INT-4536: Support Kafka Tombstones

JIRA: https://jira.spring.io/browse/INT-4536

Represent `@KafkaNull` as `null` when `@Payload(required = false)`.

spring-integration-core/src/main/java/org/springframework/integration/config/DefaultConfiguringBeanFactoryPostProcessor.java

@@ -58,6 +58,7 @@ import org.springframework.integration.handler.support.MapArgumentResolver;
 import org.springframework.integration.handler.support.PayloadExpressionArgumentResolver;
 import org.springframework.integration.handler.support.PayloadsArgumentResolver;
 import org.springframework.integration.support.DefaultMessageBuilderFactory;
+import org.springframework.integration.support.NullAwarePayloadArgumentResolver;
 import org.springframework.integration.support.SmartLifecycleRoleController;
 import org.springframework.integration.support.converter.ConfigurableCompositeMessageConverter;
 import org.springframework.integration.support.converter.DefaultDatatypeChannelMessageConverter;
@@ -99,6 +100,7 @@ class DefaultConfiguringBeanFactoryPostProcessor implements BeanFactoryPostProce
 		this.classLoader = classLoader;
 	}
 
+	@Override
 	public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException {
 		if (beanFactory instanceof BeanDefinitionRegistry) {
 			this.beanFactory = beanFactory;
@@ -477,9 +479,13 @@ class DefaultConfiguringBeanFactoryPostProcessor implements BeanFactoryPostProce
 		}
 	}
 
-	private static BeanDefinition internalArgumentResolversBuilder(boolean listCapable) {
+	private BeanDefinition internalArgumentResolversBuilder(boolean listCapable) {
 		ManagedList<BeanDefinition> resolvers = new ManagedList<>();
 		resolvers.add(new RootBeanDefinition(PayloadExpressionArgumentResolver.class));
+		BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(NullAwarePayloadArgumentResolver.class);
+		builder.addConstructorArgReference(IntegrationContextUtils.ARGUMENT_RESOLVER_MESSAGE_CONVERTER_BEAN_NAME);
+		// TODO Validator ?
+		resolvers.add(builder.getBeanDefinition());
 		resolvers.add(new RootBeanDefinition(PayloadsArgumentResolver.class));
 		resolvers.add(new RootBeanDefinition(MapArgumentResolver.class));
 

spring-integration-core/src/main/java/org/springframework/integration/handler/support/MessagingMethodInvokerHelper.java

@@ -74,6 +74,8 @@ import org.springframework.integration.annotation.ServiceActivator;
 import org.springframework.integration.annotation.UseSpelInvoker;
 import org.springframework.integration.context.IntegrationContextUtils;
 import org.springframework.integration.support.MutableMessage;
+import org.springframework.integration.support.NullAwarePayloadArgumentResolver;
+import org.springframework.integration.support.converter.ConfigurableCompositeMessageConverter;
 import org.springframework.integration.support.json.JsonObjectMapper;
 import org.springframework.integration.support.json.JsonObjectMapperProvider;
 import org.springframework.integration.util.AbstractExpressionEvaluator;
@@ -538,6 +540,19 @@ public class MessagingMethodInvokerHelper<T> extends AbstractExpressionEvaluator
 	 * that don't run in an application context.
 	 */
 	private void configureLocalMessageHandlerFactory() {
+		MessageConverter messageConverter = null;
+		if (getBeanFactory() != null &&
+				getBeanFactory()
+						.containsBean(IntegrationContextUtils.ARGUMENT_RESOLVER_MESSAGE_CONVERTER_BEAN_NAME)) {
+			messageConverter = getBeanFactory()
+									.getBean(IntegrationContextUtils.ARGUMENT_RESOLVER_MESSAGE_CONVERTER_BEAN_NAME,
+											MessageConverter.class);
+			this.messageHandlerMethodFactory.setMessageConverter(messageConverter);
+		}
+		else {
+			messageConverter = new ConfigurableCompositeMessageConverter();
+		}
+		NullAwarePayloadArgumentResolver nullResolver = new NullAwarePayloadArgumentResolver(messageConverter);
 		PayloadExpressionArgumentResolver payloadExpressionArgumentResolver = new PayloadExpressionArgumentResolver();
 		payloadExpressionArgumentResolver.setBeanFactory(getBeanFactory());
 
@@ -549,6 +564,7 @@ public class MessagingMethodInvokerHelper<T> extends AbstractExpressionEvaluator
 
 		List<HandlerMethodArgumentResolver> customArgumentResolvers = new LinkedList<>();
 		customArgumentResolvers.add(payloadExpressionArgumentResolver);
+		customArgumentResolvers.add(nullResolver);
 		customArgumentResolvers.add(payloadsArgumentResolver);
 
 		if (this.canProcessMessageList) {
@@ -560,15 +576,6 @@ public class MessagingMethodInvokerHelper<T> extends AbstractExpressionEvaluator
 		customArgumentResolvers.add(mapArgumentResolver);
 
 		this.messageHandlerMethodFactory.setCustomArgumentResolvers(customArgumentResolvers);
-
-		if (getBeanFactory() != null &&
-				getBeanFactory()
-						.containsBean(IntegrationContextUtils.ARGUMENT_RESOLVER_MESSAGE_CONVERTER_BEAN_NAME)) {
-			this.messageHandlerMethodFactory
-					.setMessageConverter(getBeanFactory()
-							.getBean(IntegrationContextUtils.ARGUMENT_RESOLVER_MESSAGE_CONVERTER_BEAN_NAME,
-									MessageConverter.class));
-		}
 	}
 
 	@SuppressWarnings("unchecked")

spring-integration-core/src/main/java/org/springframework/integration/support/NullAwarePayloadArgumentResolver.java

@@ -0,0 +1,47 @@
+/*
+ * Copyright 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.
+ * 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.integration.support;
+
+import org.springframework.messaging.converter.MessageConverter;
+import org.springframework.messaging.handler.annotation.Payload;
+import org.springframework.messaging.handler.annotation.support.PayloadArgumentResolver;
+import org.springframework.validation.Validator;
+
+/**
+ * A {@link PayloadArgumentResolver} that treats KafkaNull payloads as null.
+ * {@link Payload @Paylaod} annotation must have required = false.
+ *
+ * @author Gary Russell
+ * @since 5.1
+ *
+ */
+public class NullAwarePayloadArgumentResolver extends PayloadArgumentResolver {
+
+	public NullAwarePayloadArgumentResolver(MessageConverter messageConverter) {
+		super(messageConverter, null, false);
+	}
+
+	public NullAwarePayloadArgumentResolver(MessageConverter messageConverter, Validator validator) {
+		super(messageConverter, validator, false);
+	}
+
+	@Override
+	protected boolean isEmptyPayload(Object payload) {
+		return super.isEmptyPayload(payload) || "KafkaNull".equals(payload.getClass().getSimpleName());
+	}
+
+}

spring-integration-core/src/test/java/org/springframework/integration/endpoint/ServiceActivatorMethodResolutionTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2010 the original author or authors.
+ * Copyright 2002-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.
@@ -30,10 +30,12 @@ import org.springframework.integration.gateway.RequestReplyExchanger;
 import org.springframework.integration.handler.ServiceActivatingHandler;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.PollableChannel;
+import org.springframework.messaging.handler.annotation.Payload;
 import org.springframework.messaging.support.GenericMessage;
 
 /**
  * @author Mark Fisher
+ * @author Gary Russell
  */
 public class ServiceActivatorMethodResolutionTests {
 
@@ -231,6 +233,17 @@ public class ServiceActivatorMethodResolutionTests {
 		assertNotEquals("FOO", outputChannel.receive(10).getPayload());
 	}
 
+	@Test
+	public void nullOk() {
+		NullOkTestBean testBean = new NullOkTestBean();
+		ServiceActivatingHandler serviceActivator = new ServiceActivatingHandler(testBean);
+		QueueChannel outputChannel = new QueueChannel();
+		serviceActivator.setOutputChannel(outputChannel);
+		serviceActivator.handleMessage(new GenericMessage<>(new KafkaNull()));
+		Message<?> result = outputChannel.receive(0);
+		assertEquals("gotNull", result.getPayload());
+	}
+
 
 	@SuppressWarnings("unused")
 	private static class SingleAnnotationTestBean {
@@ -305,4 +318,32 @@ public class ServiceActivatorMethodResolutionTests {
 
 	}
 
+
+	@SuppressWarnings("unused")
+	private static class NullOkTestBean {
+
+		NullOkTestBean() {
+			super();
+		}
+
+		@ServiceActivator
+		public String nullOK(@Payload(required = false) String s) {
+			if (s == null) {
+				return "gotNull";
+			}
+			else {
+				return s;
+			}
+		}
+
+	}
+
+	private static class KafkaNull {
+
+		KafkaNull() {
+			super();
+		}
+
+	}
+
 }