From 2d8a094cf0c903c0e1466ec1d172ddbd33acde5e Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Mon, 25 Feb 2019 13:14:48 -0500 Subject: [PATCH] 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. --- ...kaListenerAnnotationBeanPostProcessor.java | 20 ++++ ...hMessagingMessageListenerAdapterTests.java | 97 +++++++++++++++++++ 2 files changed, 117 insertions(+) create mode 100644 spring-kafka/src/test/java/org/springframework/kafka/listener/adapter/BatchMessagingMessageListenerAdapterTests.java diff --git a/spring-kafka/src/main/java/org/springframework/kafka/annotation/KafkaListenerAnnotationBeanPostProcessor.java b/spring-kafka/src/main/java/org/springframework/kafka/annotation/KafkaListenerAnnotationBeanPostProcessor.java index 0c5077ed..ed25a1bc 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/annotation/KafkaListenerAnnotationBeanPostProcessor.java +++ b/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 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; diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/adapter/BatchMessagingMessageListenerAdapterTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/adapter/BatchMessagingMessageListenerAdapterTests.java new file mode 100644 index 00000000..27d3bab8 --- /dev/null +++ b/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 adapter = + (BatchMessagingMessageListenerAdapter) 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 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 kafkaListenerContainerFactory() { + ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory(); + factory.setConsumerFactory(consumerFactory()); + factory.setBatchListener(true); + return factory; + } + } + +}