SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-2662: Kafka binder DLQ root cause message (#2812)

Browse Source 
- Becasue NestedRuntimeException from Spring Framework core 6.x removed
   the getMessage method that included the detailMessage with cause in it,
   the Kafka binder DLQ records no longer include the cause message.
   Fix this issue by including the cause in the exception message.

See this comment for more details:
 - https://github.com/spring-cloud/spring-cloud-stream/issues/2662#issuecomment-1722849892
This fix is based on the following Spring Kafka commit.
 - 6f585058a6

Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/2662
This commit is contained in:
Soby Chacko
2023-09-20 11:28:13 -04:00
committed by GitHub
parent ffabdb8bf9
commit e06d4eedfd
2 changed files with 23 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java
View File

@@ -1228,7 +1228,7 @@ public class KafkaMessageChannelBinder extends
if (message.getPayload() instanceof Throwable throwablePayload) {
throwable = throwablePayload;
String exceptionMessage = buildMessage(throwable, throwable.getCause());
HeaderMode headerMode = properties.getHeaderMode();
if (headerMode == null || HeaderMode.headers.equals(headerMode)) {
@@ -1248,7 +1248,6 @@ public class KafkaMessageChannelBinder extends
.getBytes(StandardCharsets.UTF_8)));
kafkaHeaders.add(new RecordHeader(X_EXCEPTION_FQCN, throwable
.getClass().getName().getBytes(StandardCharsets.UTF_8)));
String exceptionMessage = throwable.getMessage();
if (exceptionMessage != null) {
kafkaHeaders.add(new RecordHeader(X_EXCEPTION_MESSAGE,
exceptionMessage.getBytes(StandardCharsets.UTF_8)));
@@ -1271,8 +1270,7 @@ public class KafkaMessageChannelBinder extends
record.timestampType().toString());
messageValues.put(X_EXCEPTION_FQCN,
throwable.getClass().getName());
messageValues.put(X_EXCEPTION_MESSAGE,
throwable.getMessage());
messageValues.put(X_EXCEPTION_MESSAGE, exceptionMessage);
messageValues.put(X_EXCEPTION_STACKTRACE,
getStackTraceAsString(throwable));
@@ -1323,6 +1321,26 @@ public class KafkaMessageChannelBinder extends
}
}
@Nullable
private String buildMessage(Throwable exception, Throwable cause) {
String message = exception.getMessage();
if (!exception.equals(cause)) {
if (message != null) {
message = message + "; ";
}
String causeMsg = cause.getMessage();
if (causeMsg != null) {
if (message != null) {
message = message + causeMsg;
}
else {
message = causeMsg;
}
}
}
return message;
}
@SuppressWarnings("unchecked")
@Nullable
private KafkaAwareTransactionManager<byte[], byte[]> transactionManager(@Nullable String beanName) {

2
binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java
View File

@@ -1300,7 +1300,7 @@ public class KafkaBinderTests extends
else {
assertThat(new String((byte[]) receivedMessage.getHeaders()
.get(KafkaMessageChannelBinder.X_EXCEPTION_MESSAGE))).startsWith(
"Dispatcher failed to deliver Message");
"Dispatcher failed to deliver Message; fail");
}
assertThat(receivedMessage.getHeaders()