GH-2006 Add test for KafkaNull

Resolves #2006

spring-cloud-stream/pom.xml

@@ -76,6 +76,11 @@
 			<artifactId>spring-boot-starter-web</artifactId>
 			<scope>test</scope>
 		</dependency>
+		<dependency>
+			<groupId>org.springframework.kafka</groupId>
+			<artifactId>spring-kafka</artifactId>
+			<scope>test</scope>
+		</dependency>
 	</dependencies>
 
 	<build>

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

@@ -112,7 +112,7 @@ class SmartMessageMethodArgumentResolver extends MessageMethodArgumentResolver {
 			return !StringUtils.hasText((String) payload);
 		}
 		else {
-			return false;
+			return "org.springframework.kafka.support.KafkaNull".equals(payload.getClass().getName());
 		}
 	}
 

spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java

@@ -55,6 +55,7 @@ import org.springframework.integration.dsl.IntegrationFlows;
 import org.springframework.integration.handler.LoggingHandler;
 import org.springframework.integration.scheduling.PollerMetadata;
 import org.springframework.integration.support.MessageBuilder;
+import org.springframework.kafka.support.KafkaNull;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.support.GenericMessage;
@@ -142,6 +143,26 @@ public class ImplicitFunctionBindingTests {
 		}
 	}
 
+	@Test
+	public void testNullMessage() {
+
+		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
+				TestChannelBinderConfiguration.getCompleteConfiguration(NullMessagerConfiguration.class))
+						.web(WebApplicationType.NONE)
+						.run("--spring.jmx.enabled=false", "--spring.cloud.function.definition=func")) {
+
+			InputDestination inputDestination = context.getBean(InputDestination.class);
+			OutputDestination outputDestination = context.getBean(OutputDestination.class);
+
+
+			Message inputMessage = MessageBuilder.withPayload(KafkaNull.INSTANCE).build();
+			inputDestination.send(inputMessage);
+
+			Message<byte[]> outputMessage = outputDestination.receive();
+			assertThat(outputMessage.getPayload()).isEqualTo("NULL".getBytes());
+		}
+	}
+
 	@Test
 	public void testSimpleFunctionWithStreamProperty() {
 
@@ -910,6 +931,18 @@ public class ImplicitFunctionBindingTests {
 		}
 	}
 
+	@EnableAutoConfiguration
+	public static class NullMessagerConfiguration {
+
+		@Bean
+		public Function<Person, String> func() {
+			return value -> {
+				assertThat(value).isNull();
+				return "NULL";
+			};
+		}
+	}
+
 	@EnableAutoConfiguration
 	public static class SingleReactiveConsumerConfiguration {