GH-11183: Fix batch mode message ID generation in KafkaMessageDrivenChannelAdapter

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,17 +145,22 @@ public KafkaMessageDrivenChannelAdapter(AbstractMessageListenerContainer<K, V> 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();
 			headerMapper.addTrustedPackages(
 					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);
 		}
 	}
 

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<Integer, String> 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<Integer, String> producerFactory() {
 			Map<String, Object> props = KafkaTestUtils.producerProps(this.embeddedKafkaBrokers);