dendi-wang/thingsboard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Small refactoring of sendErrorResponse

Browse Source
This commit is contained in:
Andrii Landiak 2025-06-25 16:59:31 +03:00
parent bf46d974ab
commit a3a5fd3180
1 changed files with 2 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
common/queue/src/main/java/org/thingsboard/server/queue/common/PartitionedQueueResponseTemplate.java
View File

@ -131,7 +131,7 @@ public class PartitionedQueueResponseTemplate<Request extends TbQueueMsg, Respon
@Override @Override
public void onFailure(Throwable t) { public void onFailure(Throwable t) {
log.error("[{}] Failed to send response {}", requestId, response, t); log.error("[{}] Failed to send response {}", requestId, response, t);
sendErrorResponse(requestId, responseTopic, request, t); sendErrorResponse(requestId, tpi, request, t);
stats.incrementFailed(); stats.incrementFailed();
} }
}); });
@ -158,12 +158,11 @@ public class PartitionedQueueResponseTemplate<Request extends TbQueueMsg, Respon
consumer.commit(); consumer.commit();
} }
private void sendErrorResponse(UUID requestId, String responseTopic, Request request, Throwable cause) { private void sendErrorResponse(UUID requestId, TopicPartitionInfo tpi, Request request, Throwable cause) {
Response errorResponseMsg = handler.constructErrorResponseMsg(request, cause); Response errorResponseMsg = handler.constructErrorResponseMsg(request, cause);
if (errorResponseMsg != null) { if (errorResponseMsg != null) {
errorResponseMsg.getHeaders().put(REQUEST_ID_HEADER, uuidToBytes(requestId)); errorResponseMsg.getHeaders().put(REQUEST_ID_HEADER, uuidToBytes(requestId));
TopicPartitionInfo tpi = TopicPartitionInfo.builder().topic(responseTopic).build();
responseProducer.send(tpi, errorResponseMsg, null); responseProducer.send(tpi, errorResponseMsg, null);
} }
} }