Changing the order of calling customizer

In the Kafka Streams binder, StreamsBuilderFactoryBean customzier was being called
prematurely before the object is created. Fixing this issue.

Add a test to verify

spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java

@@ -214,18 +214,16 @@ public abstract class AbstractKafkaStreamsBinderProcessor implements Application
 
 		KafkaStreamsConfiguration kafkaStreamsConfiguration = new KafkaStreamsConfiguration(streamConfigGlobalProperties);
 
-		StreamsBuilderFactoryBean streamsBuilder = this.cleanupConfig == null
+		StreamsBuilderFactoryBean streamsBuilderFactoryBean = this.cleanupConfig == null
 				? new StreamsBuilderFactoryBean(kafkaStreamsConfiguration)
 				: new StreamsBuilderFactoryBean(kafkaStreamsConfiguration,
 				this.cleanupConfig);
-		if (customizer != null) {
-			customizer.configure(streamsBuilder);
-		}
-		streamsBuilder.setAutoStartup(false);
+
+		streamsBuilderFactoryBean.setAutoStartup(false);
 		BeanDefinition streamsBuilderBeanDefinition = BeanDefinitionBuilder
 				.genericBeanDefinition(
-						(Class<StreamsBuilderFactoryBean>) streamsBuilder.getClass(),
-						() -> streamsBuilder)
+						(Class<StreamsBuilderFactoryBean>) streamsBuilderFactoryBean.getClass(),
+						() -> streamsBuilderFactoryBean)
 				.getRawBeanDefinition();
 		((BeanDefinitionRegistry) beanFactory).registerBeanDefinition(
 				"stream-builder-" + beanNamePostPrefix, streamsBuilderBeanDefinition);
@@ -234,8 +232,14 @@ public abstract class AbstractKafkaStreamsBinderProcessor implements Application
 		//Removing the application ID from global properties so that the next function won't re-use it and cause race conditions.
 		streamConfigGlobalProperties.remove(StreamsConfig.APPLICATION_ID_CONFIG);
 
-		return applicationContext.getBean(
+		final StreamsBuilderFactoryBean streamsBuilderFactoryBeanFromContext = applicationContext.getBean(
 				"&stream-builder-" + beanNamePostPrefix, StreamsBuilderFactoryBean.class);
+		//At this point, the StreamsBuilderFactoryBean is created. If the users call, getObject()
+		//in the customizer, that should grant access to the StreamsBuilder.
+		if (customizer != null) {
+			customizer.configure(streamsBuilderFactoryBean);
+		}
+		return streamsBuilderFactoryBeanFromContext;
 	}
 
 	protected Serde<?> getValueSerde(String inboundName, KafkaStreamsConsumerProperties kafkaStreamsConsumerProperties, ResolvableType resolvableType) {

spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBinderWordCountFunctionTests.java

@@ -19,6 +19,8 @@ package org.springframework.cloud.stream.binder.kafka.streams.function;
 import java.util.Arrays;
 import java.util.Date;
 import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
 
 import io.micrometer.core.instrument.MeterRegistry;
@@ -27,9 +29,9 @@ import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.common.serialization.Serdes;
 import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.kstream.Grouped;
 import org.apache.kafka.streams.kstream.KStream;
 import org.apache.kafka.streams.kstream.Materialized;
-import org.apache.kafka.streams.kstream.Serialized;
 import org.apache.kafka.streams.kstream.TimeWindows;
 import org.apache.kafka.streams.processor.StreamPartitioner;
 import org.junit.AfterClass;
@@ -44,12 +46,14 @@ import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
 import org.springframework.cloud.stream.binder.kafka.streams.InteractiveQueryService;
 import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.context.annotation.Bean;
+import org.springframework.kafka.config.StreamsBuilderFactoryBeanCustomizer;
 import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
 import org.springframework.kafka.core.DefaultKafkaProducerFactory;
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.test.EmbeddedKafkaBroker;
 import org.springframework.kafka.test.rule.EmbeddedKafkaRule;
 import org.springframework.kafka.test.utils.KafkaTestUtils;
+import org.springframework.util.Assert;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
@@ -63,6 +67,8 @@ public class KafkaStreamsBinderWordCountFunctionTests {
 
 	private static Consumer<String, String> consumer;
 
+	private final static CountDownLatch LATCH = new CountDownLatch(1);
+
 	@BeforeClass
 	public static void setUp() {
 		Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("group", "false",
@@ -100,6 +106,7 @@ public class KafkaStreamsBinderWordCountFunctionTests {
 			final MeterRegistry meterRegistry = context.getBean(MeterRegistry.class);
 			Thread.sleep(100);
 			assertThat(meterRegistry.get("stream.metrics.commit.total").gauge().value()).isEqualTo(1.0);
+			Assert.isTrue(LATCH.await(5, TimeUnit.SECONDS), "Failed to call customizers");
 		}
 	}
 
@@ -239,7 +246,7 @@ public class KafkaStreamsBinderWordCountFunctionTests {
 			return input -> input
 					.flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+")))
 					.map((key, value) -> new KeyValue<>(value, value))
-					.groupByKey(Serialized.with(Serdes.String(), Serdes.String()))
+					.groupByKey(Grouped.with(Serdes.String(), Serdes.String()))
 					.windowedBy(TimeWindows.of(5000))
 					.count(Materialized.as("foo-WordCounts"))
 					.toStream()
@@ -247,6 +254,23 @@ public class KafkaStreamsBinderWordCountFunctionTests {
 							new Date(key.window().start()), new Date(key.window().end()))));
 		}
 
+		@Bean
+		public StreamsBuilderFactoryBeanCustomizer customizer() {
+			return fb -> {
+				try {
+					fb.setStateListener((newState, oldState) -> {
+
+					});
+					fb.getObject(); //make sure no exception is thrown at this call.
+					KafkaStreamsBinderWordCountFunctionTests.LATCH.countDown();
+
+				}
+				catch (Exception e) {
+					//Nothing to do - When the exception is thrown above, the latch won't be count down.
+				}
+			};
+		}
+
 		@Bean
 		public StreamPartitioner<String, WordCount> streamPartitioner() {
 			return (t, k, v, n) -> k.equals("foo") ? 0 : 1;