GH-2249: Batch Listener LISTENER_INFO Headers

Resolves #2249

* Single String parameter for batch listeners.

* Polish docs for an empty batch.

**cherry-pick to 2.9.x, 2.8.x**

Author: garyrussell

spring-kafka-docs/src/main/asciidoc/kafka.adoc

@@ -5541,6 +5541,30 @@ When used in a `RecordInterceptor` or `RecordFilterStrategy` implementation, the
 
 The header mappers also convert to `String` when creating `MessageHeaders` from the consumer record and never map this header on an outbound record.
 
+For POJO batch listeners, starting with version 2.8.6, the header is copied into each member of the batch and is also available as a single `String` parameter after conversion.
+
+====
+[source, java]
+----
+@KafkaListener(id = "list2", topics = "someTopic", containerFactory = "batchFactory",
+        info = "info for batch")
+public void listen(List<Thing> list,
+        @Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) List<Integer> keys,
+        @Header(KafkaHeaders.RECEIVED_PARTITION_ID) List<Integer> partitions,
+        @Header(KafkaHeaders.RECEIVED_TOPIC) List<String> topics,
+        @Header(KafkaHeaders.OFFSET) List<Long> offsets,
+        @Header(KafkaHeaders.LISTENER_INFO) String info) {
+            ...
+}
+----
+====
+
+NOTE: If the batch listener has a filter and the filter results in an empty batch, you will need to add `required = false` to the `@Header` parameter because the info is not available for an empty batch.
+
+If you receive `List<Message<Thing>>` the info is in the `KafkaHeaders.LISTENER_INFO` header of each `Message<?>`.
+
+See <<batch-listeners>> for more information about consuming batches.
+
 [[dead-letters]]
 ===== Publishing Dead-letter Records
 

spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java

@@ -2269,6 +2269,9 @@ private void invokeBatchOnMessageWithRecordsOrList(final ConsumerRecords<K, V> r
 
 			ConsumerRecords<K, V> records = recordsArg;
 			List<ConsumerRecord<K, V>> recordList = recordListArg;
+			if (this.listenerinfo != null) {
+				records.iterator().forEachRemaining(rec -> listenerInfo(rec));
+			}
 			if (this.batchInterceptor != null) {
 				records = this.batchInterceptor.intercept(recordsArg, this.consumer);
 				if (records == null) {
@@ -2516,10 +2519,14 @@ private void internalHeaders(final ConsumerRecord<K, V> record) {
 				record.headers().add(new RecordHeader(KafkaHeaders.DELIVERY_ATTEMPT, buff));
 			}
 			if (this.listenerinfo != null) {
-				record.headers().add(this.infoHeader);
+				listenerInfo(record);
 			}
 		}
 
+		private void listenerInfo(final ConsumerRecord<K, V> record) {
+			record.headers().add(this.infoHeader);
+		}
+
 		private void handleNack(final ConsumerRecords<K, V> records, final ConsumerRecord<K, V> record) {
 			if (!this.autoCommit && !this.isRecordAck) {
 				processCommits();

spring-kafka/src/main/java/org/springframework/kafka/support/converter/BatchMessagingMessageConverter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2021 the original author or authors.
+ * Copyright 2016-2022 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.
@@ -169,8 +169,8 @@ public Message<?> toMessage(List<ConsumerRecord<?, ?>> records, @Nullable Acknow
 		commonHeaders(acknowledgment, consumer, rawHeaders, keys, topics, partitions, offsets, timestampTypes,
 				timestamps);
 		rawHeaders.put(KafkaHeaders.CONVERSION_FAILURES, conversionFailures);
-
 		boolean logged = false;
+		String info = null;
 		for (ConsumerRecord<?, ?> record : records) {
 			payloads.add(obtainPayload(type, record, conversionFailures));
 			keys.add(record.key());
@@ -185,6 +185,10 @@ public Message<?> toMessage(List<ConsumerRecord<?, ?>> records, @Nullable Acknow
 				Map<String, Object> converted = new HashMap<>();
 				this.headerMapper.toHeaders(record.headers(), converted);
 				convertedHeaders.add(converted);
+				Object object = converted.get(KafkaHeaders.LISTENER_INFO);
+				if (object instanceof String) {
+					info = (String) object;
+				}
 			}
 			else {
 				if (!logged) {
@@ -200,6 +204,9 @@ public Message<?> toMessage(List<ConsumerRecord<?, ?>> records, @Nullable Acknow
 				raws.add(record);
 			}
 		}
+		if (info != null) {
+			rawHeaders.put(KafkaHeaders.LISTENER_INFO, info);
+		}
 		return MessageBuilder.createMessage(payloads, kafkaMessageHeaders);
 	}
 

spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java

@@ -591,6 +591,7 @@ public void testBatchWitHeaders() throws Exception {
 		list = this.listener.offsets;
 		assertThat(list.size()).isGreaterThan(0);
 		assertThat(list.get(0)).isInstanceOf(Long.class);
+		assertThat(this.listener.listenerInfo).isEqualTo("info for batch");
 	}
 
 	@Test
@@ -1982,17 +1983,20 @@ public void listen10(List<String> list, @Header(KafkaHeaders.GROUP_ID) String gr
 			this.latch10.countDown();
 		}
 
-		@KafkaListener(id = "list2", topics = "annotated15", containerFactory = "batchFactory")
+		@KafkaListener(id = "list2", topics = "annotated15", containerFactory = "batchFactory",
+				info = "info for batch")
 		public void listen11(List<String> list,
 				@Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) List<Integer> keys,
 				@Header(KafkaHeaders.RECEIVED_PARTITION_ID) List<Integer> partitions,
 				@Header(KafkaHeaders.RECEIVED_TOPIC) List<String> topics,
-				@Header(KafkaHeaders.OFFSET) List<Long> offsets) {
+				@Header(KafkaHeaders.OFFSET) List<Long> offsets,
+				@Header(KafkaHeaders.LISTENER_INFO) String info) {
 			this.payload = list;
 			this.keys = keys;
 			this.partitions = partitions;
 			this.topics = topics;
 			this.offsets = offsets;
+			this.listenerInfo = info;
 			this.latch11.countDown();
 		}