GH-531: Support tombstones on input/output

GH-531: Set test timeout to 10 seconds

GH-531: Polishing - Fix @StreamListener

Added test for `@StreamListener`.

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;
+	}
+
+}

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(

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();
+		}
+
+	}
+
+}