From 95fce11b0a57dab70216554e322184de87799d10 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 2 Jun 2021 15:12:17 -0400 Subject: [PATCH] GH-3573: Add ConsumerRecord(s) to ConversionExcep. Resolves https://github.com/spring-projects/spring-integration/issues/3573 --- build.gradle | 2 +- .../integration/kafka/inbound/KafkaInboundGateway.java | 2 +- .../inbound/KafkaMessageDrivenChannelAdapter.java | 6 ++++-- .../kafka/inbound/MessageDrivenAdapterTests.java | 10 ++++++---- 4 files changed, 12 insertions(+), 8 deletions(-) diff --git a/build.gradle b/build.gradle index d73c7b892c..6caca8e94d 100644 --- a/build.gradle +++ b/build.gradle @@ -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' diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaInboundGateway.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaInboundGateway.java index b4b173987e..2aa42ad9f4 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaInboundGateway.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaInboundGateway.java @@ -287,7 +287,7 @@ public class KafkaInboundGateway 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) { diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java index 7cee0d9982..c25ac04eb3 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java @@ -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 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 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)); diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java index 8029039bda..f0f6e98441 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java @@ -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(); }