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

GH-3573: Add ConsumerRecord(s) to ConversionExcep.

Browse Source 
Resolves https://github.com/spring-projects/spring-integration/issues/3573
This commit is contained in:
Gary Russell
2021-06-02 15:12:17 -04:00
committed by Artem Bilan
parent f290c340f3
commit 95fce11b0a
4 changed files with 12 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
build.gradle
View File

@@ -100,7 +100,7 @@ ext {
soapVersion = '1.4.0'
springAmqpVersion = project.hasProperty('springAmqpVersion') ? project.springAmqpVersion : '2.3.7'
springDataVersion = project.hasProperty('springDataVersion') ? project.springDataVersion : '2021.0.1'
springKafkaVersion = '2.7.1'
springKafkaVersion = '2.7.2-SNAPSHOT'
springRetryVersion = '1.3.1'
springSecurityVersion = project.hasProperty('springSecurityVersion') ? project.springSecurityVersion : '5.5.0'
springVersion = project.hasProperty('springVersion') ? project.springVersion : '5.3.7'

2
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaInboundGateway.java
View File

@@ -287,7 +287,7 @@ public class KafkaInboundGateway<K, V, R> extends MessagingGatewaySupport implem
catch (RuntimeException e) {
if (getErrorChannel() != null) {
KafkaInboundGateway.this.messagingTemplate.send(getErrorChannel(), buildErrorMessage(null,
new ConversionException("Failed to convert to message for: " + record, e)));
new ConversionException("Failed to convert to message", record, e)));
}
}
if (message != null) {

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

@@ -21,6 +21,7 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.BiConsumer;
import java.util.stream.Collectors;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
@@ -449,7 +450,7 @@ public class KafkaMessageDrivenChannelAdapter<K, V> extends MessageProducerSuppo
setAttributesIfNecessary(record, message);
}
catch (RuntimeException e) {
RuntimeException exception = new ConversionException("Failed to convert to message for: " + record, e);
RuntimeException exception = new ConversionException("Failed to convert to message", record, e);
sendErrorMessageIfNecessary(null, exception);
}
@@ -541,7 +542,8 @@ public class KafkaMessageDrivenChannelAdapter<K, V> extends MessageProducerSuppo
setAttributesIfNecessary(records, message);
}
catch (RuntimeException e) {
Exception exception = new ConversionException("Failed to convert to message for: " + records, e);
Exception exception = new ConversionException("Failed to convert to message",
records.stream().collect(Collectors.toList()), e);
MessageChannel errorChannel = getErrorChannel();
if (errorChannel != null) {
getMessagingTemplate().send(errorChannel, new ErrorMessage(exception));

10
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2020 the original author or authors.
* Copyright 2016-2021 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -217,7 +217,8 @@ class MessageDrivenAdapterTests {
assertThat(error).isNotNull();
assertThat(error.getPayload()).isInstanceOf(ConversionException.class);
assertThat(((ConversionException) error.getPayload()).getMessage())
.contains("Failed to convert to message for: ConsumerRecord(topic = testTopic1");
.contains("Failed to convert to message");
assertThat(((ConversionException) error.getPayload()).getRecord()).isNotNull();
adapter.stop();
}
@@ -462,13 +463,14 @@ class MessageDrivenAdapterTests {
});
PollableChannel errors = new QueueChannel();
adapter.setErrorChannel(errors);
template.sendDefault(1, "foo");
template.sendDefault(1, "bar");
Message<?> error = errors.receive(10000);
assertThat(error).isNotNull();
assertThat(error.getPayload()).isInstanceOf(ConversionException.class);
assertThat(((ConversionException) error.getPayload()).getMessage())
.contains("Failed to convert to message for: [ConsumerRecord(topic = testTopic2");
.contains("Failed to convert to message");
assertThat(((ConversionException) error.getPayload()).getRecords()).hasSize(2);
adapter.stop();
}