Add KafkaNull converter, upgrade to s-c-function 3.0.10

pom.xml

@@ -25,7 +25,7 @@
 		<java.version>1.8</java.version>
 		<reactor.version>Dysprosium-SR6</reactor.version>
 		<objenesis.version>2.1</objenesis.version>
-		<spring-cloud-function.version>3.0.10.BUILD-SNAPSHOT</spring-cloud-function.version>
+		<spring-cloud-function.version>3.0.10.RELEASE</spring-cloud-function.version>
 		<maven-checkstyle-plugin.failsOnError>true</maven-checkstyle-plugin.failsOnError>
 		<maven-checkstyle-plugin.failsOnViolation>true</maven-checkstyle-plugin.failsOnViolation>
 		<maven-checkstyle-plugin.includeTestSourceDirectory>true</maven-checkstyle-plugin.includeTestSourceDirectory>

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderFactoryAutoConfiguration.java

@@ -20,6 +20,7 @@ import java.io.IOException;
 import java.net.URL;
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.Enumeration;
 import java.util.HashMap;
 import java.util.LinkedList;
@@ -59,13 +60,18 @@ import org.springframework.integration.handler.support.PayloadExpressionArgument
 import org.springframework.integration.handler.support.PayloadsArgumentResolver;
 import org.springframework.integration.support.NullAwarePayloadArgumentResolver;
 import org.springframework.lang.Nullable;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.messaging.converter.AbstractMessageConverter;
 import org.springframework.messaging.converter.CompositeMessageConverter;
+import org.springframework.messaging.converter.MessageConverter;
 import org.springframework.messaging.handler.annotation.support.DefaultMessageHandlerMethodFactory;
 import org.springframework.messaging.handler.annotation.support.HeaderMethodArgumentResolver;
 import org.springframework.messaging.handler.annotation.support.HeadersMethodArgumentResolver;
 import org.springframework.messaging.handler.annotation.support.MessageHandlerMethodFactory;
 import org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolver;
 import org.springframework.util.ClassUtils;
+import org.springframework.util.MimeTypeUtils;
 import org.springframework.util.StringUtils;
 import org.springframework.validation.Validator;
 
@@ -112,7 +118,10 @@ public class BinderFactoryAutoConfiguration {
 		return parsedBinderConfigurations;
 	}
 
-
+	@Bean
+	public MessageConverter kafkaNullConverter() {
+		return new KafkaNullConverter();
+	}
 
 	@Bean(IntegrationContextUtils.MESSAGE_HANDLER_FACTORY_BEAN_NAME)
 	public static MessageHandlerMethodFactory messageHandlerMethodFactory(
@@ -236,4 +245,33 @@ public class BinderFactoryAutoConfiguration {
 		return new CompositeMessageChannelConfigurer(configurerList);
 	}
 
+	static class KafkaNullConverter extends AbstractMessageConverter {
+
+		KafkaNullConverter() {
+			super(Collections.singletonList(MimeTypeUtils.ALL));
+		}
+
+		@Override
+		protected boolean supports(Class<?> aClass) {
+			return "org.springframework.kafka.support.KafkaNull".equals(aClass.getName());
+		}
+
+		@Override
+		protected boolean canConvertFrom(Message<?> message, Class<?> targetClass) {
+			return message.getPayload().getClass().getName().equals("org.springframework.kafka.support.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;
+		}
+
+	}
 }