GH-1386: Fix StreamsBuilder customization

Resolves #1386

The builder was incorrectly customized in `start()`. It needs to be
customized after creation so that when it is used to create `KStream`s
elsewhere, it is already customized.

Author: garyrussell

spring-kafka/src/main/java/org/springframework/kafka/config/StreamsBuilderFactoryBean.java

@@ -224,7 +224,9 @@ protected synchronized StreamsBuilder createInstance() {
 			Assert.state(this.properties != null,
 					"streams configuration properties must not be null");
 		}
-		return new StreamsBuilder();
+		StreamsBuilder builder = new StreamsBuilder();
+		this.infrastructureCustomizer.configureBuilder(builder);
+		return builder;
 	}
 
 	@Override
@@ -246,9 +248,7 @@ public synchronized void start() {
 			try {
 				Assert.state(this.properties != null,
 						"streams configuration properties must not be null");
-				StreamsBuilder builder = getObject();
-				this.infrastructureCustomizer.configureBuilder(builder);
-				Topology topology = builder.build(this.properties); // NOSONAR: getObject() cannot return null
+				Topology topology = getObject().build(this.properties); // NOSONAR: getObject() cannot return null
 				this.infrastructureCustomizer.configureTopology(topology);
 				LOGGER.debug(() -> topology.describe().toString());
 				this.kafkaStreams = new KafkaStreams(topology, this.properties, this.clientSupplier);

spring-kafka/src/test/java/org/springframework/kafka/config/KafkaStreamsCustomizerTests.java

@@ -25,16 +25,21 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.common.serialization.Serdes;
 import org.apache.kafka.streams.KafkaStreams;
+import org.apache.kafka.streams.KeyValue;
 import org.apache.kafka.streams.StreamsBuilder;
 import org.apache.kafka.streams.StreamsConfig;
 import org.apache.kafka.streams.Topology;
 import org.apache.kafka.streams.errors.DeserializationExceptionHandler;
+import org.apache.kafka.streams.kstream.KStream;
+import org.apache.kafka.streams.kstream.Transformer;
 import org.apache.kafka.streams.processor.ProcessorContext;
+import org.apache.kafka.streams.state.StoreBuilder;
+import org.apache.kafka.streams.state.Stores;
 import org.junit.jupiter.api.Test;
 
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.kafka.annotation.EnableKafka;
@@ -45,7 +50,6 @@
 import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
 
-
 /**
  * @author Nurettin Yilmaz
  * @author Artem Bilan
@@ -75,11 +79,11 @@ public void testKafkaStreamsCustomizer(@Autowired KafkaStreamsConfiguration conf
 		assertThat(STATE_LISTENER.getCurrentState()).isEqualTo(state);
 		Properties properties = configuration.asProperties();
 		assertThat(properties.get(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG))
-			.isEqualTo(Collections.singletonList(config.brokerAddresses));
+				.isEqualTo(Collections.singletonList(config.broker.getBrokersAsString()));
 		assertThat(properties.get(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG))
 				.isEqualTo(Foo.class);
 		assertThat(properties.get(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG))
-			.isEqualTo(1000);
+				.isEqualTo(1000);
 		assertThat(this.config.builderConfigured.get()).isTrue();
 		assertThat(this.config.topologyConfigured.get()).isTrue();
 	}
@@ -93,20 +97,24 @@ public static class KafkaStreamsConfig {
 
 		final AtomicBoolean topologyConfigured = new AtomicBoolean();
 
-		@Value("${" + EmbeddedKafkaBroker.SPRING_EMBEDDED_KAFKA_BROKERS + "}")
-		private String brokerAddresses;
+		@Autowired
+		EmbeddedKafkaBroker broker;
 
 		@Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_BUILDER_BEAN_NAME)
 		public StreamsBuilderFactoryBean defaultKafkaStreamsBuilder() {
-			StreamsBuilderFactoryBean streamsBuilderFactoryBean =
-					new StreamsBuilderFactoryBean(kStreamsConfigs());
+			StreamsBuilderFactoryBean streamsBuilderFactoryBean = new StreamsBuilderFactoryBean(kStreamsConfigs());
 			streamsBuilderFactoryBean.setKafkaStreamsCustomizer(customizer());
 			streamsBuilderFactoryBean.setInfrastructureCustomizer(new KafkaStreamsInfrastructureCustomizer() {
 
 				@SuppressWarnings("unchecked")
 				@Override
 				public void configureBuilder(StreamsBuilder builder) {
 					KafkaStreamsConfig.this.builderConfigured.set(true);
+					StoreBuilder<?> storeBuilder = Stores.keyValueStoreBuilder(
+							Stores.persistentKeyValueStore("testStateStore"),
+							Serdes.Integer(),
+							Serdes.String());
+					builder.addStateStore(storeBuilder);
 				}
 
 				@Override
@@ -122,17 +130,40 @@ public void configureTopology(Topology topology) {
 		public KafkaStreamsConfiguration kStreamsConfigs() {
 			Map<String, Object> props = new HashMap<>();
 			props.put(StreamsConfig.APPLICATION_ID_CONFIG, APPLICATION_ID);
-			props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, Collections.singletonList(this.brokerAddresses));
+			props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,
+					Collections.singletonList(this.broker.getBrokersAsString()));
 			props.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, Foo.class);
 			props.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, 1000);
 			return new KafkaStreamsConfiguration(props);
 		}
 
-
 		private KafkaStreamsCustomizer customizer() {
 			return kafkaStreams -> kafkaStreams.setStateListener(STATE_LISTENER);
 		}
 
+		@Bean
+		public KStream<String, String> testStream(StreamsBuilder kStreamBuilder) {
+			KStream<String, String> stream = kStreamBuilder.stream("test_topic");
+
+			stream
+					.transform(() -> new Transformer<String, String, KeyValue<String, String>>() {
+						@Override
+						public void init(ProcessorContext context) {
+						}
+
+						@Override
+						public KeyValue<String, String> transform(String key, String value) {
+							return null;
+						}
+
+						@Override
+						public void close() {
+						}
+					}, "testStateStore")
+					.to("test_output");
+
+			return stream;
+		}
 	}
 
 	static class TestStateListener implements KafkaStreams.StateListener {