From 723eadb67f9997a9bd29c4ac87a90115cc75f5b5 Mon Sep 17 00:00:00 2001
From: Oleg Zhurakousky <ozhurakousky@vmware.com>
Date: Wed, 8 Jun 2022 20:06:06 +0200
Subject: [PATCH] GH-2355 Add tests for KafkaNull batching suport

This commit also has some checksyle fixes due to s-c-build changes
Inspired by suggestion from https://github.com/spring-cloud/spring-cloud-stream/pull/2422 by @sbernardo

Resolves #2355
---
 .../binder/test/FunctionBindingTestUtils.java     |  6 +++++-
 .../test/TestChannelBinderConfiguration.java      |  3 +++
 .../stream/function/FunctionBatchingTests.java    | 15 +++++++++++----
 .../function/ImplicitFunctionBindingTests.java    |  1 -
 4 files changed, 19 insertions(+), 6 deletions(-)

diff --git a/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/test/FunctionBindingTestUtils.java b/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/test/FunctionBindingTestUtils.java
index ff671a482..384f33a3d 100644
--- a/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/test/FunctionBindingTestUtils.java
+++ b/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/test/FunctionBindingTestUtils.java
@@ -42,7 +42,11 @@ import org.springframework.messaging.MessageChannel;
  * @since 3.0.2
  *
  */
-public class FunctionBindingTestUtils {
+public final class FunctionBindingTestUtils {
+
+	private FunctionBindingTestUtils() {
+
+	}
 
 	@SuppressWarnings("rawtypes")
 	public static void bind(ConfigurableApplicationContext applicationContext, Object function) {
diff --git a/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/test/TestChannelBinderConfiguration.java b/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/test/TestChannelBinderConfiguration.java
index d13d61a3c..61ba1487c 100644
--- a/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/test/TestChannelBinderConfiguration.java
+++ b/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/test/TestChannelBinderConfiguration.java
@@ -48,6 +48,9 @@ import org.springframework.integration.config.EnableIntegration;
 @EnableIntegration
 public class TestChannelBinderConfiguration<T> {
 
+	/**
+	 * The name of the test binder.
+	 */
 	public static final String NAME = "integration";
 
 	/**
diff --git a/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/FunctionBatchingTests.java b/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/FunctionBatchingTests.java
index 396a8d4e3..14a4cd355 100644
--- a/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/FunctionBatchingTests.java
+++ b/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/FunctionBatchingTests.java
@@ -29,6 +29,7 @@ import org.springframework.cloud.stream.binder.test.OutputDestination;
 import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
 import org.springframework.context.annotation.Bean;
 import org.springframework.integration.support.MessageBuilder;
+import org.springframework.kafka.support.KafkaNull;
 import org.springframework.messaging.Message;
 
 import static org.assertj.core.api.Assertions.assertThat;
@@ -44,7 +45,7 @@ import static org.assertj.core.api.Assertions.assertThat;
 public class FunctionBatchingTests {
 
 	@Test
-	void testMessageBatchConfiguration() {
+	void testMessageBatchConfigurationWithKafkaNull() {
 		TestChannelBinderConfiguration.applicationContextRunner(MessageBatchConfiguration.class)
 				.withPropertyValues("spring.jmx.enabled=false",
 						"spring.cloud.stream.function.definition=func",
@@ -54,10 +55,12 @@ public class FunctionBatchingTests {
 					OutputDestination outputDestination = context
 							.getBean(OutputDestination.class);
 
-					List<byte[]> list = new ArrayList<>();
+					List<Object> list = new ArrayList<>();
 					list.add("{\"name\":\"bob\"}".getBytes());
 					list.add("{\"name\":\"jill\"}".getBytes());
-					Message<List<byte[]>> inputMessage = MessageBuilder
+					list.add(KafkaNull.INSTANCE);
+					list.add("{\"name\":\"steve\"}".getBytes());
+					Message<List<Object>> inputMessage = MessageBuilder
 							.withPayload(list)
 							.build();
 					inputDestination.send(inputMessage);
@@ -282,7 +285,11 @@ public class FunctionBatchingTests {
 
 		@Bean
 		public Function<Message<List<Person>>, Person> func() {
-			return x -> x.getPayload().get(0);
+			return x -> {
+				Object o = x.getPayload().get(2);
+				assertThat(o).isInstanceOf(KafkaNull.class);
+				return (Person) x.getPayload().get(0);
+			};
 		}
 
 		public static class Person {
diff --git a/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java b/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java
index f221c957b..b3a70b35d 100644
--- a/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java
+++ b/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java
@@ -1428,7 +1428,6 @@ public class ImplicitFunctionBindingTests {
 		@Bean
 		public Function<Message<List<Map<String, String>>>, Message<List<Map<String, String>>>> funcB() {
 			return v -> {
-				System.out.println(v);
 				return v;
 			};
 		}