GH-2971 Properly handle Kafka tombstone payload conversion

See the corresponding commit in s-c-function
Resolves #2971

core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/function/FunctionBatchingTests.java

@@ -45,6 +45,23 @@ import static org.assertj.core.api.Assertions.assertThat;
  */
 class FunctionBatchingTests {
 
+	@Test
+	void messageWithKafkaNull() {
+		TestChannelBinderConfiguration.applicationContextRunner(KafkaNullConfiguration.class)
+				.withPropertyValues("spring.cloud.stream.function.definition=myFunction").run(context -> {
+					InputDestination inputDestination = context.getBean(InputDestination.class);
+					OutputDestination outputDestination = context.getBean(OutputDestination.class);
+
+					var message = MessageBuilder.withPayload(KafkaNull.INSTANCE).build();
+					inputDestination.send(message);
+
+					Object kn = outputDestination.receive().getPayload();
+
+					assertThat(kn).isInstanceOf(KafkaNull.class);
+					context.stop();
+				});
+	}
+
 	@Test
 	void messageBatchConfigurationWithKafkaNull() {
 		TestChannelBinderConfiguration.applicationContextRunner(MessageBatchConfiguration.class)
@@ -300,4 +317,12 @@ class FunctionBatchingTests {
 
 	}
 
+	@EnableAutoConfiguration
+	public static class KafkaNullConfiguration {
+		@Bean
+		public Function<Message<?>, Message<?>> myFunction() {
+			return v -> MessageBuilder.withPayload(KafkaNull.INSTANCE).build();
+		}
+	}
+
 }