GH-974 Batch listeners replace KafkaNull with null

Fixes https://github.com/spring-projects/spring-kafka/issues/974

Replace `KafkaNull` elements with `null` when the payload is resolved to a list.

spring-kafka/src/main/java/org/springframework/kafka/annotation/KafkaListenerAnnotationBeanPostProcessor.java

@@ -56,6 +56,7 @@ import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
 import org.springframework.beans.factory.config.Scope;
 import org.springframework.context.expression.StandardBeanExpressionResolver;
 import org.springframework.core.MethodIntrospector;
+import org.springframework.core.MethodParameter;
 import org.springframework.core.Ordered;
 import org.springframework.core.annotation.AnnotatedElementUtils;
 import org.springframework.core.annotation.AnnotationUtils;
@@ -73,6 +74,7 @@ import org.springframework.kafka.config.MultiMethodKafkaListenerEndpoint;
 import org.springframework.kafka.listener.KafkaListenerErrorHandler;
 import org.springframework.kafka.support.KafkaNull;
 import org.springframework.kafka.support.TopicPartitionInitialOffset;
+import org.springframework.messaging.Message;
 import org.springframework.messaging.converter.GenericMessageConverter;
 import org.springframework.messaging.handler.annotation.support.DefaultMessageHandlerMethodFactory;
 import org.springframework.messaging.handler.annotation.support.HeaderMethodArgumentResolver;
@@ -832,6 +834,24 @@ public class KafkaListenerAnnotationBeanPostProcessor<K, V>
 			argumentResolvers.add(new MessageMethodArgumentResolver(messageConverter));
 			argumentResolvers.add(new PayloadArgumentResolver(messageConverter, validator) {
 
+
+				@Override
+				public Object resolveArgument(MethodParameter parameter, Message<?> message) throws Exception {
+					Object resolved = super.resolveArgument(parameter, message);
+					/*
+					 * Replace KafkaNull list elements with null.
+					 */
+					if (resolved instanceof List) {
+						List<?> list = ((List<?>) resolved);
+						for (int i = 0; i < list.size(); i++) {
+							if (list.get(i) instanceof KafkaNull) {
+								list.set(i, null);
+							}
+						}
+					}
+					return resolved;
+				}
+
 				@Override
 				protected boolean isEmptyPayload(Object payload) {
 					return payload == null || payload instanceof KafkaNull;

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

@@ -0,0 +1,97 @@
+/*
+ * Copyright 2019 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.kafka.listener.adapter;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+
+import java.util.Collections;
+import java.util.List;
+
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.junit.jupiter.api.Test;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.annotation.EnableKafka;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
+import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
+import org.springframework.kafka.core.ConsumerFactory;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
+
+/**
+ * @author Gary Russell
+ * @since 2.2.5
+ *
+ */
+@SpringJUnitConfig
+@DirtiesContext
+public class BatchMessagingMessageListenerAdapterTests {
+
+	@SuppressWarnings("unchecked")
+	@Test
+	public void testKafkaNullInList(@Autowired KafkaListenerEndpointRegistry registry, @Autowired Foo foo) {
+		BatchMessagingMessageListenerAdapter<String, String> adapter =
+				(BatchMessagingMessageListenerAdapter<String, String>) registry
+					.getListenerContainer("foo").getContainerProperties().getMessageListener();
+		adapter.onMessage(Collections.singletonList(new ConsumerRecord<>("foo", 0, 0L, null, null)), null, null);
+		assertThat(foo.value).isNull();
+	}
+
+	public static class Foo {
+
+		public String value = "someValue";
+
+		@KafkaListener(id = "foo", topics = "foo", autoStartup = "false")
+		public void listen(List<String> list) {
+			list.forEach(s -> {
+				this.value = s;
+			});
+		}
+
+	}
+
+	@Configuration
+	@EnableKafka
+	public static class Config {
+
+		@Bean
+		public Foo foo() {
+			return new Foo();
+		}
+
+		@SuppressWarnings({ "rawtypes" })
+		@Bean
+		public ConsumerFactory consumerFactory() {
+			ConsumerFactory consumerFactory = mock(ConsumerFactory.class);
+			return consumerFactory;
+		}
+
+		@SuppressWarnings({ "rawtypes", "unchecked" })
+		@Bean
+		public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
+			ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();
+			factory.setConsumerFactory(consumerFactory());
+			factory.setBatchListener(true);
+			return factory;
+		}
+	}
+
+}