GH-548: StreamsBuilderFactoryBean enhancements

Fixes https://github.com/spring-projects/spring-kafka/issues/548

* Make `StreamsConfig` customizable in the `StreamsBuilderFactoryBean`
* Set the phase on `StreamsBuilderFactoryBean` to the `Integer.MAX_VALUE - 1000`
* Adding tests

* Addressing PR review comments

* Addressing PR review comments

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * Copyright 2017-2018 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -36,6 +36,7 @@ import org.springframework.util.Assert;
  *
  * @author Artem Bilan
  * @author Ivan Ursul
+ * @author Soby Chacko
  *
  * @since 1.1.4
  */
@@ -43,17 +44,17 @@ public class StreamsBuilderFactoryBean extends AbstractFactoryBean<StreamsBuilde
 
 	private static final int DEFAULT_CLOSE_TIMEOUT = 10;
 
-	private final StreamsConfig streamsConfig;
-
 	private final CleanupConfig cleanupConfig;
 
+	private StreamsConfig streamsConfig;
+
 	private KafkaStreams kafkaStreams;
 
 	private KafkaClientSupplier clientSupplier = new DefaultKafkaClientSupplier();
 
 	private boolean autoStartup = true;
 
-	private int phase = Integer.MIN_VALUE;
+	private int phase = Integer.MAX_VALUE - 1000;
 
 	private KafkaStreams.StateListener stateListener;
 
@@ -63,6 +64,16 @@ public class StreamsBuilderFactoryBean extends AbstractFactoryBean<StreamsBuilde
 
 	private volatile boolean running;
 
+	/**
+	 * Default constructor that creates the factory without a {@link StreamsConfig}.
+	 * It is the factory user's responsibility to properly set {@link StreamsConfig}
+	 * using {@link StreamsBuilderFactoryBean#setStreamsConfig(StreamsConfig)}
+	 * @since 2.1.3.
+	 */
+	public StreamsBuilderFactoryBean() {
+		this.cleanupConfig = new CleanupConfig();
+	}
+
 	/**
 	 * Construct an instance with the supplied streams configuration.
 	 * @param streamsConfig the streams configuration.
@@ -107,6 +118,16 @@ public class StreamsBuilderFactoryBean extends AbstractFactoryBean<StreamsBuilde
 		this.cleanupConfig = cleanupConfig;
 	}
 
+	/**
+	 * Set {@link StreamsConfig} on this factory.
+	 * @param streamsConfig the streams configuration.
+	 * @since 2.1.3
+	 */
+	public void setStreamsConfig(StreamsConfig streamsConfig) {
+		Assert.notNull(streamsConfig, "'streamsConfig' must not be null");
+		this.streamsConfig = streamsConfig;
+	}
+
 	public void setClientSupplier(KafkaClientSupplier clientSupplier) {
 		Assert.notNull(clientSupplier, "'clientSupplier' must not be null");
 		this.clientSupplier = clientSupplier; // NOSONAR (sync)
@@ -137,6 +158,9 @@ public class StreamsBuilderFactoryBean extends AbstractFactoryBean<StreamsBuilde
 
 	@Override
 	protected StreamsBuilder createInstance() throws Exception {
+		if (this.autoStartup) {
+			Assert.notNull(this.streamsConfig, "'streamsConfig' must not be null");
+		}
 		return new StreamsBuilder();
 	}
 
@@ -166,6 +190,7 @@ public class StreamsBuilderFactoryBean extends AbstractFactoryBean<StreamsBuilde
 	public synchronized void start() {
 		if (!this.running) {
 			try {
+				Assert.notNull(this.streamsConfig, "'streamsConfig' must not be null");
 				this.kafkaStreams = new KafkaStreams(getObject().build(), this.streamsConfig, this.clientSupplier);
 				this.kafkaStreams.setStateListener(this.stateListener);
 				this.kafkaStreams.setUncaughtExceptionHandler(this.exceptionHandler);

spring-kafka/src/test/java/org/springframework/kafka/core/StreamsBuilderFactoryLateConfigTests.java

@@ -0,0 +1,95 @@
+/*
+ * Copyright 2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.kafka.core;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.kafka.streams.StreamsConfig;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+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.KafkaException;
+import org.springframework.kafka.annotation.EnableKafka;
+import org.springframework.kafka.annotation.EnableKafkaStreams;
+import org.springframework.kafka.annotation.KafkaStreamsDefaultConfiguration;
+import org.springframework.kafka.test.context.EmbeddedKafka;
+import org.springframework.kafka.test.rule.KafkaEmbedded;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.junit4.SpringRunner;
+
+/**
+ * @author Soby Chacko
+ */
+@RunWith(SpringRunner.class)
+@DirtiesContext
+@EmbeddedKafka
+public class StreamsBuilderFactoryLateConfigTests {
+
+	private static final String APPLICATION_ID = "streamsBuilderFactoryLateConfigTests";
+
+	@Value("${" + KafkaEmbedded.SPRING_EMBEDDED_KAFKA_BROKERS + "}")
+	private String brokerAddresses;
+
+	@Autowired
+	private StreamsBuilderFactoryBean streamsBuilderFactoryBean;
+
+	@Test(expected = KafkaException.class)
+	public void testStreamBuilderFactoryCannotBeStartedWithoutStreamconfig() {
+		StreamsBuilderFactoryBean streamsBuilderFactoryBean = new StreamsBuilderFactoryBean();
+		streamsBuilderFactoryBean.start();
+	}
+
+	@Test(expected = IllegalArgumentException.class)
+	public void testStreamBuilderFactoryCannotBeInstantiatedWhenAutoStart() throws Exception {
+		StreamsBuilderFactoryBean streamsBuilderFactoryBean = new StreamsBuilderFactoryBean();
+		streamsBuilderFactoryBean.setAutoStartup(true);
+		streamsBuilderFactoryBean.createInstance();
+	}
+
+	@Test
+	public void testStreamsBuilderFactoryWithConfigProvidedLater() {
+		Map<String, Object> props = new HashMap<>();
+		props.put(StreamsConfig.APPLICATION_ID_CONFIG, APPLICATION_ID);
+		props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, this.brokerAddresses);
+		StreamsConfig streamsConfig = new StreamsConfig(props);
+		streamsBuilderFactoryBean.setStreamsConfig(streamsConfig);
+
+		assertThat(streamsBuilderFactoryBean.isRunning()).isFalse();
+		streamsBuilderFactoryBean.start();
+		assertThat(streamsBuilderFactoryBean.isRunning()).isTrue();
+	}
+
+	@Configuration
+	@EnableKafka
+	@EnableKafkaStreams
+	public static class KafkaStreamsConfiguration {
+
+		@Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_BUILDER_BEAN_NAME)
+		public StreamsBuilderFactoryBean defaultKafkaStreamsBuilder() {
+			StreamsBuilderFactoryBean streamsBuilderFactoryBean = new StreamsBuilderFactoryBean();
+			streamsBuilderFactoryBean.setAutoStartup(false);
+			return streamsBuilderFactoryBean;
+		}
+	}
+}