Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix KafkaMDChannelAdapter error handling

Browse Source 
Current `IntegrationRecordMessageListener.onMessage()`
send a conversion error to the `errorChannel`, but
it does not return if it was successful leading to
the NPE `enhanceHeadersAndSaveAttributes()` because the `message`
is null

* Change the `IntegrationRecordMessageListener.onMessage()` logic to check
the result of the `sendErrorMessageIfNecessary()` and return immediately
if success.
Rethrow an exception otherwise.
This commit is contained in:
Artem Bilan
2022-05-31 14:08:07 -04:00
parent 7174e8840b
commit cdcc986d11
1 changed files with 7 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java
View File

@@ -443,8 +443,7 @@ public class KafkaMessageDrivenChannelAdapter<K, V> extends MessageProducerSuppo
@Override
public void onMessage(ConsumerRecord<K, V> record, Acknowledgment acknowledgment, Consumer<?, ?> consumer) {
Message<?> message = null;
Message<?> message;
try {
message = toMessagingMessage(record, acknowledgment, consumer);
}
@@ -452,22 +451,22 @@ public class KafkaMessageDrivenChannelAdapter<K, V> extends MessageProducerSuppo
if (KafkaMessageDrivenChannelAdapter.this.retryTemplate == null) {
setAttributesIfNecessary(record, null, true);
}
MessageChannel errorChannel = getErrorChannel();
if (errorChannel != null) {
RuntimeException exception = new ConversionException("Failed to convert to message", record, ex);
sendErrorMessageIfNecessary(null, exception);
RuntimeException exception = new ConversionException("Failed to convert to message", record, ex);
if (sendErrorMessageIfNecessary(null, exception)) {
return;
}
else {
throw ex;
}
}
RetryTemplate template = KafkaMessageDrivenChannelAdapter.this.retryTemplate;
if (template != null) {
Message<?> toSend = message;
doWithRetry(template, KafkaMessageDrivenChannelAdapter.this.recoveryCallback, record, acknowledgment,
consumer, () -> {
if (!KafkaMessageDrivenChannelAdapter.this.filterInRetry || passesFilter(record)) {
sendMessageIfAny(enhanceHeadersAndSaveAttributes(toSend, record), record);
sendMessageIfAny(enhanceHeadersAndSaveAttributes(message, record), record);
}
});
}