GH-601: @KafkaListener: Support List<Message<Foo>>

Resolves #601

Author: garyrussell

Diff for: spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/BatchMessagingMessageListenerAdapter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2017 the original author or authors.
+ * Copyright 2016-2018 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.
@@ -61,7 +61,7 @@ public class BatchMessagingMessageListenerAdapter<K, V> extends MessagingMessage
 
 	private static final Message<KafkaNull> NULL_MESSAGE = new GenericMessage<>(KafkaNull.INSTANCE);
 
-	private BatchMessageConverter messageConverter = new BatchMessagingMessageConverter();
+	private BatchMessageConverter batchMessageConverter = new BatchMessagingMessageConverter();
 
 	private KafkaListenerErrorHandler errorHandler;
 
@@ -79,7 +79,10 @@ public BatchMessagingMessageListenerAdapter(Object bean, Method method, KafkaLis
 	 * @param messageConverter the converter.
 	 */
 	public void setBatchMessageConverter(BatchMessageConverter messageConverter) {
-		this.messageConverter = messageConverter;
+		this.batchMessageConverter = messageConverter;
+		if (messageConverter.getRecordMessageConverter() != null) {
+			setMessageConverter(messageConverter.getRecordMessageConverter());
+		}
 	}
 
 	/**
@@ -89,7 +92,7 @@ public void setBatchMessageConverter(BatchMessageConverter messageConverter) {
 	 * being able to convert {@link org.springframework.messaging.Message}.
 	 */
 	protected final BatchMessageConverter getBatchMessageConverter() {
-		return this.messageConverter;
+		return this.batchMessageConverter;
 	}
 
 	/**

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

@@ -428,9 +428,12 @@ else if (parameterizedType.getRawType().equals(List.class)
 							this.isConsumerRecordList =	paramType.equals(ConsumerRecord.class)
 									|| (paramType instanceof ParameterizedType
 										&& ((ParameterizedType) paramType).getRawType().equals(ConsumerRecord.class));
-							this.isMessageList = paramType.equals(Message.class)
-									|| (paramType instanceof ParameterizedType
-											&& ((ParameterizedType) paramType).getRawType().equals(Message.class));
+							boolean messageHasGeneric = paramType instanceof ParameterizedType
+									&& ((ParameterizedType) paramType).getRawType().equals(Message.class);
+							this.isMessageList = paramType.equals(Message.class) || messageHasGeneric;
+							if (messageHasGeneric) {
+								genericParameterType = ((ParameterizedType) paramType).getActualTypeArguments()[0];
+							}
 						}
 					}
 				}

Diff for: spring-kafka/src/main/java/org/springframework/kafka/support/converter/BatchMessageConverter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016 the original author or authors.
+ * Copyright 2016-2016 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.
@@ -24,6 +24,7 @@
 import org.apache.kafka.clients.producer.ProducerRecord;
 
 import org.springframework.kafka.support.Acknowledgment;
+import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 
 /**
@@ -53,4 +54,15 @@ Message<?> toMessage(List<ConsumerRecord<?, ?>> records, Acknowledgment acknowle
 	 */
 	List<ProducerRecord<?, ?>> fromMessage(Message<?> message, String defaultTopic);
 
+	/**
+	 * Return the record converter used by this batch converter, if configured,
+	 * or null.
+	 * @return the converter or null.
+	 * @since 2.1.5
+	 */
+	@Nullable
+	default RecordMessageConverter getRecordMessageConverter() {
+		return null;
+	}
+
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2017 the original author or authors.
+ * Copyright 2016-2018 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.
@@ -117,6 +117,11 @@ public void setHeaderMapper(KafkaHeaderMapper headerMapper) {
 		this.headerMapper = headerMapper;
 	}
 
+	@Override
+	public RecordMessageConverter getRecordMessageConverter() {
+		return this.recordConverter;
+	}
+
 	@Override
 	public Message<?> toMessage(List<ConsumerRecord<?, ?>> records, Acknowledgment acknowledgment,
 			Consumer<?, ?> consumer, Type type) {

Diff for: spring-kafka/src/test/java/org/springframework/kafka/annotation/BatchListenerConversionTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * Copyright 2017-2018 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.
@@ -43,6 +43,7 @@
 import org.springframework.kafka.support.converter.StringJsonMessageConverter;
 import org.springframework.kafka.test.rule.KafkaEmbedded;
 import org.springframework.kafka.test.utils.KafkaTestUtils;
+import org.springframework.messaging.Message;
 import org.springframework.messaging.handler.annotation.Header;
 import org.springframework.messaging.support.GenericMessage;
 import org.springframework.test.annotation.DirtiesContext;
@@ -62,7 +63,7 @@ public class BatchListenerConversionTests {
 	private static final String DEFAULT_TEST_GROUP_ID = "blc";
 
 	@ClassRule // one topic to preserve order
-	public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, 1, "blc1", "blc2");
+	public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, 1, "blc1", "blc2", "blc3");
 
 	@Autowired
 	private Config config;
@@ -90,6 +91,20 @@ private void doTest(Listener listener, String topic) throws InterruptedException
 		assertThat((listener.receivedPartitions).get(0)).isEqualTo(0);
 	}
 
+	@Test
+	public void testBatchOfPojoMessages() throws Exception {
+		String topic = "blc3";
+		this.template.send(new GenericMessage<>(
+				new Foo("bar"), Collections.singletonMap(KafkaHeaders.TOPIC, topic)));
+		this.template.send(new GenericMessage<>(
+				new Foo("baz"), Collections.singletonMap(KafkaHeaders.TOPIC, topic)));
+		Listener3 listener = this.config.listener3();
+		assertThat(listener.latch1.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(listener.received.size()).isGreaterThan(0);
+		assertThat(listener.received.get(0).getPayload()).isInstanceOf(Foo.class);
+		assertThat(listener.received.get(0).getPayload().getBar()).isEqualTo("bar");
+	}
+
 	@Configuration
 	@EnableKafka
 	public static class Config {
@@ -149,6 +164,11 @@ public Listener listener2() {
 			return new Listener("blc2");
 		}
 
+		@Bean
+		public Listener3 listener3() {
+			return new Listener3();
+		}
+
 	}
 
 	public static class Listener {
@@ -191,6 +211,22 @@ public String getTopic() {
 
 	}
 
+	public static class Listener3 {
+
+		private final CountDownLatch latch1 = new CountDownLatch(1);
+
+		private List<Message<Foo>> received;
+
+		@KafkaListener(topics = "blc3", groupId = "blc3")
+		public void listen1(List<Message<Foo>> foos) {
+			if (this.received == null) {
+				this.received = foos;
+			}
+			this.latch1.countDown();
+		}
+
+	}
+
 	public static class Foo {
 
 		public String bar;

Diff for: src/reference/asciidoc/kafka.adoc

@@ -1,4 +1,3 @@
-
 [[kafka]]
 === Using Spring for Apache Kafka
 
@@ -863,6 +862,11 @@ public void listen15(List<Message<?>> list, Acknowledgment ack) {
 }
 ----
 
+No conversion is performed on the payloads in this case.
+
+If the `BatchMessagingMessageConverter` is configured with a `RecordMessageConverter`, you can also add a generic type to the `Message` parameter and the payloads will be converted.
+See <<payload-conversion-with-batch>> for more information.
+
 You can also receive a list of `ConsumerRecord<?, ?>` objects but it must be the only parameter (aside from an optional `Acknowledgment` when using manual commits) defined on the method:
 
 [source, java]
@@ -1382,6 +1386,8 @@ When a container is paused, it continues to `poll()` the consumer, avoiding a re
 [[serdes]]
 ==== Serialization/Deserialization and Message Conversion
 
+===== Overview
+
 Apache Kafka provides a high-level API for serializing/deserializing record values as well as their keys.
 It is present with the `org.apache.kafka.common.serialization.Serializer<T>` and
 `org.apache.kafka.common.serialization.Deserializer<T>` abstractions with some built-in implementations.
@@ -1460,7 +1466,10 @@ With a class-level `@KafkaListener`, the payload type is used to select which `@
 
 NOTE: When using the `StringJsonMessageConverter`, you should use a `StringDeserializer` in the kafka consumer configuration and `StringSerializer` in the kafka producer configuration, when using Spring Integration or the `KafkaTemplate.send(Message<?> message)` method.
 
-Starting with _version 1.3.2_ you can also use a `StringJsonMessageConverter` within a `BatchMessagingMessageConverter` for converting batch messages, when using a batch listener container factory.
+[[payload-conversion-with-batch]]
+===== Payload Conversion with Batch Listeners
+
+Starting with _version 1.3.2_, you can also use a `StringJsonMessageConverter` within a `BatchMessagingMessageConverter` for converting batch messages, when using a batch listener container factory.
 
 By default, the type for the conversion is inferred from the listener argument.
 If you configure the `StringJsonMessageConverter` with a `DefaultJackson2TypeMapper` that has its `TypePrecedence` set to `TYPE_ID` (instead of the default `INFERRED`), then the converter will use type information in headers (if present) instead.
@@ -1498,8 +1507,20 @@ public void listen(List<Foo> foos, @Header(KafkaHeaders.OFFSET) List<Long> offse
 
 Notice that you can still access the batch headers too.
 
-Starting with _versions 2.1.1_, the `org.springframework.core.convert.ConversionService` used by the default
-`o.s.messaging.handler.annotation.support.MessageHandlerMethodFactory` to reslove parameters for the invocation
+If the batch converter has a record converter that supports it, you can also receive a list of messages where the payloads are converted according to the generic type:
+
+[source, java]
+----
+@KafkaListener(topics = "blc3", groupId = "blc3")
+public void listen1(List<Message<Foo>> fooMessages) {
+    ...
+}
+----
+
+===== ConversionService Customization
+
+Starting with _version 2.1.1_, the `org.springframework.core.convert.ConversionService` used by the default
+`o.s.messaging.handler.annotation.support.MessageHandlerMethodFactory` to resolve parameters for the invocation
 of a listener method is supplied with all beans implementing any of the following interfaces:
 
  - `org.springframework.core.convert.converter.Converter`