From 4407eefa4f3838320d637efbffa39333f1df4093 Mon Sep 17 00:00:00 2001 From: zernov-a Date: Wed, 15 Jul 2026 17:47:46 +0300 Subject: [PATCH 1/3] GH-11183: Fix batch mode message ID generation in KafkaMessageDrivenChannelAdapter Fixes: gh-11183 Signed-off-by: zernov-a --- .../KafkaMessageDrivenChannelAdapter.java | 10 ++++-- .../integration/kafka/dsl/KafkaDslTests.java | 35 +++++++++++++++++++ 2 files changed, 43 insertions(+), 2 deletions(-) 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 e1f87062811..09031b018f9 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 @@ -57,6 +57,7 @@ import org.springframework.kafka.support.JsonKafkaHeaderMapper; import org.springframework.kafka.support.KafkaHeaders; import org.springframework.kafka.support.converter.BatchMessageConverter; +import org.springframework.kafka.support.converter.BatchMessagingMessageConverter; import org.springframework.kafka.support.converter.ConversionException; import org.springframework.kafka.support.converter.KafkaMessageHeaders; import org.springframework.kafka.support.converter.MessageConverter; @@ -144,7 +145,6 @@ public KafkaMessageDrivenChannelAdapter(AbstractMessageListenerContainer m .toArray(new String[0])); messageConverter.setHeaderMapper(headerMapper); this.recordListener.setMessageConverter(messageConverter); - this.batchListener.setMessageConverter(messageConverter); } else if (JacksonPresent.isJackson2Present()) { var headerMapper = new org.springframework.kafka.support.DefaultKafkaHeaderMapper(); @@ -153,7 +153,13 @@ else if (JacksonPresent.isJackson2Present()) { .toArray(new String[0])); messageConverter.setHeaderMapper(headerMapper); this.recordListener.setMessageConverter(messageConverter); - this.batchListener.setMessageConverter(messageConverter); + } + + if (this.mode.equals(ListenerMode.batch)) { + var batchMessagingMessageConverter = new BatchMessagingMessageConverter(messageConverter); + batchMessagingMessageConverter.setGenerateMessageId(true); + batchMessagingMessageConverter.setGenerateTimestamp(true); + this.batchListener.setBatchMessageConverter(batchMessagingMessageConverter); } } diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java index a79b34973c5..5daef1dc8e3 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java @@ -123,6 +123,8 @@ public class KafkaDslTests { static final String TEST_TOPIC8 = "test-topic8"; + static final String TEST_TOPIC10 = "test-topic10"; + @Autowired @Qualifier("sendToKafkaFlow.input") private MessageChannel sendToKafkaFlowInput; @@ -133,6 +135,9 @@ public class KafkaDslTests { @Autowired private PollableChannel listeningFromKafkaResults2; + @Autowired + private PollableChannel listeningFromKafkaResults3; + @Autowired @Qualifier("kafkaProducer1.handler") private KafkaProducerMessageHandler kafkaProducer1; @@ -155,6 +160,9 @@ public class KafkaDslTests { @Qualifier("kafkaTemplate:" + TEST_TOPIC2) private KafkaTemplate kafkaTemplateTopic2; + @Autowired + private KafkaTemplate kafkaTemplate; + @Autowired private JsonKafkaHeaderMapper mapper; @@ -195,6 +203,7 @@ void testKafkaAdapters() throws Exception { assertThat(headers.containsKey(KafkaHeaders.ACKNOWLEDGMENT)).isTrue(); Acknowledgment acknowledgment = headers.get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment.class); acknowledgment.acknowledge(); + assertThat(headers.get(MessageHeaders.ID)).isNotNull(); assertThat(headers.get(KafkaHeaders.RECEIVED_TOPIC)).isEqualTo(TEST_TOPIC1); assertThat(headers.get(KafkaHeaders.RECEIVED_KEY)).isEqualTo(i + 1); assertThat(headers.get(KafkaHeaders.RECEIVED_PARTITION)).isEqualTo(0); @@ -211,6 +220,7 @@ void testKafkaAdapters() throws Exception { assertThat(headers.containsKey(KafkaHeaders.ACKNOWLEDGMENT)).isTrue(); Acknowledgment acknowledgment = headers.get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment.class); acknowledgment.acknowledge(); + assertThat(headers.get(MessageHeaders.ID)).isNotNull(); assertThat(headers.get(KafkaHeaders.RECEIVED_TOPIC)).isEqualTo(TEST_TOPIC2); assertThat(headers.get(KafkaHeaders.RECEIVED_KEY)).isEqualTo(i + 1); assertThat(headers.get(KafkaHeaders.RECEIVED_PARTITION)).isEqualTo(0); @@ -269,6 +279,17 @@ void channels(@Qualifier("topic6Channel") MessageChannel topic6Channel, .isEqualTo("foo"); } + @Test + void testBatchListener() { + this.kafkaTemplate.send(TEST_TOPIC10, "foo"); + + Message receive = this.listeningFromKafkaResults3.receive(20000); + assertThat(receive).isNotNull(); + assertThat(receive.getPayload()).isEqualTo("foo"); + MessageHeaders headers = receive.getHeaders(); + assertThat(headers.get(MessageHeaders.ID)).isNotNull(); + } + @Configuration @EnableIntegration @EnableKafka @@ -341,6 +362,20 @@ public IntegrationFlow topic2ListenerFromKafkaFlow() { .get(); } + @Bean + public IntegrationFlow batchListenerFromKafkaFlow() { + return IntegrationFlow + .from(Kafka + .messageDrivenChannelAdapter(consumerFactory(), + KafkaMessageDrivenChannelAdapter.ListenerMode.batch, TEST_TOPIC10) + .configureListenerContainer(c -> + c.ackMode(ContainerProperties.AckMode.MANUAL) + .id("topic10ListenerContainer"))) + .split() + .channel(c -> c.queue("listeningFromKafkaResults3")) + .get(); + } + @Bean public ProducerFactory producerFactory() { Map props = KafkaTestUtils.producerProps(this.embeddedKafkaBrokers); From 5cb38195d3f758f0d73d4966b4c7faa200743e33 Mon Sep 17 00:00:00 2001 From: zernov-a Date: Wed, 15 Jul 2026 18:43:11 +0300 Subject: [PATCH 2/3] Add @author tags and consolidate converter setup in KafkaMessageDrivenChannelAdapter Signed-off-by: zernov-a --- .../kafka/inbound/KafkaMessageDrivenChannelAdapter.java | 5 +++-- .../springframework/integration/kafka/dsl/KafkaDslTests.java | 1 + 2 files changed, 4 insertions(+), 2 deletions(-) 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 09031b018f9..e2a7969979d 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 @@ -78,6 +78,7 @@ * @author Artem Bilan * @author Urs Keller * @author Jooyoung Pyoung + * @author Zernov Oleksii * * @since 5.4 */ @@ -144,7 +145,6 @@ public KafkaMessageDrivenChannelAdapter(AbstractMessageListenerContainer m JacksonMessagingUtils.DEFAULT_TRUSTED_PACKAGES .toArray(new String[0])); messageConverter.setHeaderMapper(headerMapper); - this.recordListener.setMessageConverter(messageConverter); } else if (JacksonPresent.isJackson2Present()) { var headerMapper = new org.springframework.kafka.support.DefaultKafkaHeaderMapper(); @@ -152,9 +152,10 @@ else if (JacksonPresent.isJackson2Present()) { org.springframework.integration.support.json.JacksonJsonUtils.DEFAULT_TRUSTED_PACKAGES .toArray(new String[0])); messageConverter.setHeaderMapper(headerMapper); - this.recordListener.setMessageConverter(messageConverter); } + this.recordListener.setMessageConverter(messageConverter); + if (this.mode.equals(ListenerMode.batch)) { var batchMessagingMessageConverter = new BatchMessagingMessageConverter(messageConverter); batchMessagingMessageConverter.setGenerateMessageId(true); diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java index 5daef1dc8e3..fe40d429a8c 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java @@ -97,6 +97,7 @@ * @author Anshul Mehra * @author Jooyoung Pyoung * @author Glenn Renfro + * @author Zernov Oleksii * * @since 5.4 */ From 9d705ddbbc179a65d55ab593f975358537b0e6e3 Mon Sep 17 00:00:00 2001 From: zernov-a Date: Wed, 15 Jul 2026 22:47:36 +0300 Subject: [PATCH 3/3] remove foo/bar and redundant test config Signed-off-by: zernov-a --- .../inbound/KafkaMessageDrivenChannelAdapter.java | 2 +- .../integration/kafka/dsl/KafkaDslTests.java | 11 ++++------- 2 files changed, 5 insertions(+), 8 deletions(-) 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 e2a7969979d..167aae7b2bf 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 @@ -78,7 +78,7 @@ * @author Artem Bilan * @author Urs Keller * @author Jooyoung Pyoung - * @author Zernov Oleksii + * @author Oleksii Zernov * * @since 5.4 */ diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java index fe40d429a8c..517b3344083 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java @@ -97,7 +97,7 @@ * @author Anshul Mehra * @author Jooyoung Pyoung * @author Glenn Renfro - * @author Zernov Oleksii + * @author Oleksii Zernov * * @since 5.4 */ @@ -282,11 +282,11 @@ void channels(@Qualifier("topic6Channel") MessageChannel topic6Channel, @Test void testBatchListener() { - this.kafkaTemplate.send(TEST_TOPIC10, "foo"); + this.kafkaTemplate.send(TEST_TOPIC10, "testData"); Message receive = this.listeningFromKafkaResults3.receive(20000); assertThat(receive).isNotNull(); - assertThat(receive.getPayload()).isEqualTo("foo"); + assertThat(receive.getPayload()).isEqualTo("testData"); MessageHeaders headers = receive.getHeaders(); assertThat(headers.get(MessageHeaders.ID)).isNotNull(); } @@ -368,10 +368,7 @@ public IntegrationFlow batchListenerFromKafkaFlow() { return IntegrationFlow .from(Kafka .messageDrivenChannelAdapter(consumerFactory(), - KafkaMessageDrivenChannelAdapter.ListenerMode.batch, TEST_TOPIC10) - .configureListenerContainer(c -> - c.ackMode(ContainerProperties.AckMode.MANUAL) - .id("topic10ListenerContainer"))) + KafkaMessageDrivenChannelAdapter.ListenerMode.batch, TEST_TOPIC10)) .split() .channel(c -> c.queue("listeningFromKafkaResults3")) .get();