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..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 @@ -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; @@ -77,6 +78,7 @@ * @author Artem Bilan * @author Urs Keller * @author Jooyoung Pyoung + * @author Oleksii Zernov * * @since 5.4 */ @@ -143,8 +145,6 @@ public KafkaMessageDrivenChannelAdapter(AbstractMessageListenerContainer m JacksonMessagingUtils.DEFAULT_TRUSTED_PACKAGES .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(); @@ -152,8 +152,15 @@ 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.batchListener.setMessageConverter(messageConverter); + } + + this.recordListener.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..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,6 +97,7 @@ * @author Anshul Mehra * @author Jooyoung Pyoung * @author Glenn Renfro + * @author Oleksii Zernov * * @since 5.4 */ @@ -123,6 +124,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 +136,9 @@ public class KafkaDslTests { @Autowired private PollableChannel listeningFromKafkaResults2; + @Autowired + private PollableChannel listeningFromKafkaResults3; + @Autowired @Qualifier("kafkaProducer1.handler") private KafkaProducerMessageHandler kafkaProducer1; @@ -155,6 +161,9 @@ public class KafkaDslTests { @Qualifier("kafkaTemplate:" + TEST_TOPIC2) private KafkaTemplate kafkaTemplateTopic2; + @Autowired + private KafkaTemplate kafkaTemplate; + @Autowired private JsonKafkaHeaderMapper mapper; @@ -195,6 +204,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 +221,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 +280,17 @@ void channels(@Qualifier("topic6Channel") MessageChannel topic6Channel, .isEqualTo("foo"); } + @Test + void testBatchListener() { + this.kafkaTemplate.send(TEST_TOPIC10, "testData"); + + Message receive = this.listeningFromKafkaResults3.receive(20000); + assertThat(receive).isNotNull(); + assertThat(receive.getPayload()).isEqualTo("testData"); + MessageHeaders headers = receive.getHeaders(); + assertThat(headers.get(MessageHeaders.ID)).isNotNull(); + } + @Configuration @EnableIntegration @EnableKafka @@ -341,6 +363,17 @@ public IntegrationFlow topic2ListenerFromKafkaFlow() { .get(); } + @Bean + public IntegrationFlow batchListenerFromKafkaFlow() { + return IntegrationFlow + .from(Kafka + .messageDrivenChannelAdapter(consumerFactory(), + KafkaMessageDrivenChannelAdapter.ListenerMode.batch, TEST_TOPIC10)) + .split() + .channel(c -> c.queue("listeningFromKafkaResults3")) + .get(); + } + @Bean public ProducerFactory producerFactory() { Map props = KafkaTestUtils.producerProps(this.embeddedKafkaBrokers);