diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaNullConverter.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaNullConverter.java new file mode 100644 index 000000000..9006b9c4e --- /dev/null +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaNullConverter.java @@ -0,0 +1,62 @@ +/* + * Copyright 2019-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.cloud.stream.binder.kafka; + +import java.util.Collections; + +import org.springframework.kafka.support.KafkaNull; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHeaders; +import org.springframework.messaging.converter.AbstractMessageConverter; +import org.springframework.messaging.converter.MessageConverter; + +/** + * A {@link MessageConverter} that supports {@link KafkaNull} payloads. + * + * @author Gary Russell + * @author Aldo Sinanaj + * @since 2.2 + */ +public class KafkaNullConverter extends AbstractMessageConverter { + + public KafkaNullConverter() { + super(Collections.emptyList()); + } + + @Override + protected boolean supports(Class aClass) { + return KafkaNull.class.equals(aClass); + } + + @Override + protected boolean canConvertFrom(Message message, Class targetClass) { + return message.getPayload() instanceof KafkaNull; + } + + @Override + protected Object convertFromInternal(Message message, Class targetClass, + Object conversionHint) { + return message.getPayload(); + } + + @Override + protected Object convertToInternal(Object payload, MessageHeaders headers, + Object conversionHint) { + return payload; + } + +} diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java index bef6264cc..c83e49cdf 100644 --- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java @@ -29,10 +29,12 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean import org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration; import org.springframework.boot.autoconfigure.kafka.KafkaProperties; import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.cloud.stream.annotation.StreamMessageConverter; import org.springframework.cloud.stream.binder.Binder; import org.springframework.cloud.stream.binder.kafka.KafkaBinderMetrics; import org.springframework.cloud.stream.binder.kafka.KafkaBindingRebalanceListener; import org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder; +import org.springframework.cloud.stream.binder.kafka.KafkaNullConverter; import org.springframework.cloud.stream.binder.kafka.properties.JaasLoginModuleConfiguration; import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties; import org.springframework.cloud.stream.binder.kafka.properties.KafkaExtendedBindingProperties; @@ -46,6 +48,7 @@ import org.springframework.kafka.security.jaas.KafkaJaasLoginModuleInitializer; import org.springframework.kafka.support.LoggingProducerListener; import org.springframework.kafka.support.ProducerListener; import org.springframework.lang.Nullable; +import org.springframework.messaging.converter.MessageConverter; /** * Kafka binder configuration class. @@ -59,6 +62,7 @@ import org.springframework.lang.Nullable; * @author Gary Russell * @author Oleg Zhurakousky * @author Artem Bilan + * @author Aldo Sinanaj */ @Configuration @ConditionalOnMissingBean(Binder.class) @@ -112,6 +116,13 @@ public class KafkaBinderConfiguration { return new LoggingProducerListener(); } + @Bean + @StreamMessageConverter + @ConditionalOnMissingBean(KafkaNullConverter.class) + MessageConverter kafkaNullConverter() { + return new KafkaNullConverter(); + } + @Bean @ConditionalOnMissingBean(KafkaJaasLoginModuleInitializer.class) public KafkaJaasLoginModuleInitializer jaasInitializer( diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaNullConverterTest.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaNullConverterTest.java new file mode 100644 index 000000000..3e3c99b22 --- /dev/null +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaNullConverterTest.java @@ -0,0 +1,122 @@ +/* + * Copyright 2016-2017 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.cloud.stream.binder.kafka.integration; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.ClassRule; +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.boot.test.context.TestConfiguration; +import org.springframework.cloud.stream.annotation.EnableBinding; +import org.springframework.cloud.stream.annotation.StreamListener; +import org.springframework.cloud.stream.messaging.Processor; +import org.springframework.kafka.annotation.KafkaListener; +import org.springframework.kafka.support.KafkaNull; +import org.springframework.kafka.test.rule.EmbeddedKafkaRule; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.handler.annotation.Payload; +import org.springframework.messaging.support.GenericMessage; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.junit4.SpringRunner; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * @author Aldo Sinanaj + * @author Gary Russell + */ +@RunWith(SpringRunner.class) +@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE, + properties = "spring.cloud.stream.binding.input.destination=output") +@EnableBinding(Processor.class) +@DirtiesContext +public class KafkaNullConverterTest { + + private static final String KAFKA_BROKERS_PROPERTY = "spring.kafka.bootstrap-servers"; + + @Autowired + private MessageChannel output; + + @Autowired + private MessageChannel input; + + @Autowired + private KafkaNullConverterTestConfig config; + + @ClassRule + public static EmbeddedKafkaRule kafkaEmbedded = new EmbeddedKafkaRule(1, true); + + @BeforeClass + public static void setup() { + System.setProperty(KAFKA_BROKERS_PROPERTY, + kafkaEmbedded.getEmbeddedKafka().getBrokersAsString()); + } + + @AfterClass + public static void clean() { + System.clearProperty(KAFKA_BROKERS_PROPERTY); + } + + @Test + public void testKafkaNullConverterOutput() throws InterruptedException { + this.output.send(new GenericMessage<>(KafkaNull.INSTANCE)); + + assertThat(this.config.countDownLatchOutput.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(this.config.outputPayload).isNull(); + } + + @Test + public void testKafkaNullConverterInput() throws InterruptedException { + this.input.send(new GenericMessage<>(KafkaNull.INSTANCE)); + + assertThat(this.config.countDownLatchInput.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(this.config.inputPayload).isNull(); + } + + @TestConfiguration + public static class KafkaNullConverterTestConfig { + + final CountDownLatch countDownLatchOutput = new CountDownLatch(1); + + final CountDownLatch countDownLatchInput = new CountDownLatch(1); + + volatile byte[] outputPayload = new byte[0]; + + volatile byte[] inputPayload = new byte[0]; + + @KafkaListener(id = "foo", topics = "output") + public void listen(@Payload(required = false) byte[] in) { + this.outputPayload = in; + countDownLatchOutput.countDown(); + } + + @StreamListener(Processor.INPUT) + public void inputListen(@Payload(required = false) byte[] payload) { + this.inputPayload = payload; + countDownLatchInput.countDown(); + } + + } + +}