Fix BatchToRecordAdapter

Incorrectly passed the whole message list to the callback, which
caused listeners with a `ConsumerRecord<?, ?>` argument to fail.

Changes a public API but there is no choice and it was only added a couple of
weeks ago.

Author: garyrussell

spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/BatchToRecordAdapter.java

@@ -61,12 +61,12 @@ interface Callback<K, V> {
 
 		/**
 		 * Handle each message.
-		 * @param records the records.
+		 * @param record the record.
 		 * @param ack the acknowledgment.
 		 * @param consumer the consumer.
 		 * @param message the message.
 		 */
-		void invoke(List<ConsumerRecord<K, V>> records, Acknowledgment ack, Consumer<?, ?> consumer,
+		void invoke(ConsumerRecord<K, V> record, Acknowledgment ack, Consumer<?, ?> consumer,
 				Message<?> message);
 
 	}

spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/DefaultBatchToRecordAdapter.java

@@ -68,7 +68,7 @@ public void adapt(List<Message<?>> messages, List<ConsumerRecord<K, V>> records,
 		for (int i = 0; i < messages.size(); i++) {
 			Message<?> message = messages.get(i);
 			try {
-				callback.invoke(records, ack, consumer, message);
+				callback.invoke(records.get(i), ack, consumer, message);
 			}
 			catch (Exception e) {
 				this.recoverer.accept(records.get(i), e);

spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/MessagingMessageListenerAdapter.java

@@ -318,6 +318,7 @@ protected Message<?> toMessagingMessage(ConsumerRecord<K, V> record, Acknowledgm
 	 */
 	protected final Object invokeHandler(Object data, Acknowledgment acknowledgment, Message<?> message,
 			Consumer<?, ?> consumer) {
+
 		try {
 			if (data instanceof List && !this.isConsumerRecordList) {
 				return this.handlerMethod.invoke(message, acknowledgment, consumer);

spring-kafka/src/test/java/org/springframework/kafka/listener/adapter/BatchListenerWithRecordAdapterTests.java

@@ -57,22 +57,48 @@ void test(@Autowired KafkaListenerEndpointRegistry registry, @Autowired TestList
 		records.add(barRecord);
 		records.add(new ConsumerRecord<String, String>("foo", 0, 2, null, "baz"));
 		adapter.onMessage(records, null, null);
-		assertThat(foo.values).contains("foo", "bar", "baz");
+		assertThat(foo.values1).contains("foo", "bar", "baz");
 		assertThat(config.failed).isSameAs(barRecord);
 	}
 
+	@SuppressWarnings("unchecked")
+	@Test
+	void testFullRecord(@Autowired KafkaListenerEndpointRegistry registry, @Autowired TestListener foo,
+			@Autowired Config config) {
+
+		config.failed = null;
+		BatchMessagingMessageListenerAdapter<String, String> adapter =
+				(BatchMessagingMessageListenerAdapter<String, String>) registry
+					.getListenerContainer("batchRecordAdapterFullRecord").getContainerProperties().getMessageListener();
+		List<ConsumerRecord<String, String>> records = new ArrayList<>();
+		records.add(new ConsumerRecord<String, String>("foo", 0, 0, null, "foo"));
+		ConsumerRecord<String, String> barRecord = new ConsumerRecord<String, String>("foo", 0, 1, null, "bar");
+		records.add(barRecord);
+		records.add(new ConsumerRecord<String, String>("foo", 0, 2, null, "baz"));
+		adapter.onMessage(records, null, null);
+		assertThat(foo.values2).contains("foo", "bar", "baz");
+		assertThat(config.failed).isNull();
+	}
+
 	public static class TestListener {
 
-		final List<String> values = new ArrayList<>();
+		final List<String> values1 = new ArrayList<>();
+
+		final List<String> values2 = new ArrayList<>();
 
 		@KafkaListener(id = "batchRecordAdapter", topics = "foo", autoStartup = "false")
-		public void listen(String data) {
-			values.add(data);
+		public void listen1(String data) {
+			values1.add(data);
 			if ("bar".equals(data)) {
 				throw new RuntimeException("reject partial");
 			}
 		}
 
+		@KafkaListener(id = "batchRecordAdapterFullRecord", topics = "foo", autoStartup = "false")
+		public void listen2(ConsumerRecord<Integer, String> data) {
+			values2.add(data.value());
+		}
+
 	}
 
 	@Configuration