Add KafkaStreamsConfiguration

spring-kafka/src/main/java/org/springframework/kafka/annotation/KafkaStreamsDefaultConfiguration.java

@@ -16,15 +16,14 @@
 
 package org.springframework.kafka.annotation;
 
-import java.util.Properties;
-
 import org.apache.kafka.streams.StreamsConfig;
 
 import org.springframework.beans.factory.ObjectProvider;
 import org.springframework.beans.factory.UnsatisfiedDependencyException;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.config.KafkaStreamsConfiguration;
 import org.springframework.kafka.core.StreamsBuilderFactoryBean;
 
 /**
@@ -57,8 +56,10 @@ public class KafkaStreamsDefaultConfiguration {
 
 	@Bean(name = DEFAULT_STREAMS_BUILDER_BEAN_NAME)
 	public StreamsBuilderFactoryBean defaultKafkaStreamsBuilder(
-			@Qualifier(DEFAULT_STREAMS_CONFIG_BEAN_NAME) ObjectProvider<Properties> streamsConfigProvider) {
-		Properties streamsConfig = streamsConfigProvider.getIfAvailable();
+			@Qualifier(DEFAULT_STREAMS_CONFIG_BEAN_NAME)
+				ObjectProvider<KafkaStreamsConfiguration> streamsConfigProvider) {
+
+		KafkaStreamsConfiguration streamsConfig = streamsConfigProvider.getIfAvailable();
 		if (streamsConfig != null) {
 			return new StreamsBuilderFactoryBean(streamsConfig);
 		}

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

@@ -0,0 +1,66 @@
+/*
+ * 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.config;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+import org.apache.kafka.streams.StreamsBuilder;
+
+import org.springframework.util.Assert;
+
+/**
+ * Wrapper for {@link StreamsBuilder} properties.
+ *
+ * @author Gary Russell
+ * @since 2.2
+ *
+ */
+public class KafkaStreamsConfiguration {
+
+	private final Map<String, Object> configs;
+
+	private Properties properties;
+
+	public KafkaStreamsConfiguration(Map<String, Object> configs) {
+		Assert.notNull(configs, "Configuration map cannot be null");
+		this.configs = new HashMap<>(configs);
+	}
+
+	/**
+	 * Return the configuration map as a {@link Properties}.
+	 * @return the properties.
+	 */
+	public Properties asProperties() {
+		if (this.properties == null) {
+			Properties properties = new Properties();
+			this.configs.forEach((k, v) -> {
+				String value = v.toString();
+				if (v instanceof List && value.length() > 1) {
+					// trim [...] - revert to comma-delimited list
+					value = value.substring(1, value.length() - 1);
+				}
+				properties.setProperty(k, value);
+			});
+			this.properties = properties;
+		}
+		return this.properties;
+	}
+
+}

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

@@ -31,6 +31,7 @@
 import org.springframework.beans.factory.config.AbstractFactoryBean;
 import org.springframework.context.SmartLifecycle;
 import org.springframework.kafka.KafkaException;
+import org.springframework.kafka.config.KafkaStreamsConfiguration;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
@@ -93,7 +94,9 @@ public StreamsBuilderFactoryBean() {
 	/**
 	 * Construct an instance with the supplied streams configuration.
 	 * @param streamsConfig the streams configuration.
+	 * @deprecated in favor of {@link #StreamsBuilderFactoryBean(KafkaStreamsConfiguration)}
 	 */
+	@Deprecated
 	public StreamsBuilderFactoryBean(StreamsConfig streamsConfig) {
 		this(streamsConfig, new CleanupConfig());
 	}
@@ -104,7 +107,7 @@ public StreamsBuilderFactoryBean(StreamsConfig streamsConfig) {
 	 * @param streamsConfig the streams configuration.
 	 * @param cleanupConfig the cleanup configuration.
 	 * @since 2.1.2.
-	 * @deprecated in favor of {@link #StreamsBuilderFactoryBean(Properties, CleanupConfig)}
+	 * @deprecated in favor of {@link #StreamsBuilderFactoryBean(KafkaStreamsConfiguration, CleanupConfig)}
 	 */
 	@Deprecated
 	public StreamsBuilderFactoryBean(StreamsConfig streamsConfig, CleanupConfig cleanupConfig) {
@@ -121,17 +124,17 @@ public StreamsBuilderFactoryBean(StreamsConfig streamsConfig, CleanupConfig clea
 	 * @param cleanupConfig the cleanup configuration.
 	 * @since 2.2
 	 */
-	public StreamsBuilderFactoryBean(Properties streamsConfig, CleanupConfig cleanupConfig) {
+	public StreamsBuilderFactoryBean(KafkaStreamsConfiguration streamsConfig, CleanupConfig cleanupConfig) {
 		Assert.notNull(streamsConfig, "'streamsConfig' must not be null");
 		Assert.notNull(cleanupConfig, "'cleanupConfig' must not be null");
-		this.properties = streamsConfig;
+		this.properties = streamsConfig.asProperties();
 		this.cleanupConfig = cleanupConfig;
 	}
 
 	/**
 	 * Construct an instance with the supplied streams configuration.
 	 * @param streamsConfig the streams configuration.
-	 * @deprecated in favor of {@link #StreamsBuilderFactoryBean(Properties)}.
+	 * @deprecated in favor of {@link #StreamsBuilderFactoryBean(KafkaStreamsConfiguration)}.
 	 */
 	@Deprecated
 	public StreamsBuilderFactoryBean(Map<String, Object> streamsConfig) {
@@ -143,7 +146,7 @@ public StreamsBuilderFactoryBean(Map<String, Object> streamsConfig) {
 	 * @param streamsConfig the streams configuration.
 	 * @since 2.2
 	 */
-	public StreamsBuilderFactoryBean(Properties streamsConfig) {
+	public StreamsBuilderFactoryBean(KafkaStreamsConfiguration streamsConfig) {
 		this(streamsConfig, new CleanupConfig());
 	}
 
@@ -153,7 +156,7 @@ public StreamsBuilderFactoryBean(Properties streamsConfig) {
 	 * @param streamsConfig the streams configuration.
 	 * @param cleanupConfig the cleanup configuration.
 	 * @since 2.1.2.
-	 * @deprecated in favor of {@link #StreamsBuilderFactoryBean(Properties, CleanupConfig)}.
+	 * @deprecated in favor of {@link #StreamsBuilderFactoryBean(KafkaStreamsConfiguration, CleanupConfig)}.
 	 */
 	@Deprecated
 	public StreamsBuilderFactoryBean(Map<String, Object> streamsConfig, CleanupConfig cleanupConfig) {

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

@@ -18,7 +18,8 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 
-import java.util.Properties;
+import java.util.HashMap;
+import java.util.Map;
 
 import org.apache.kafka.streams.KafkaStreams;
 import org.apache.kafka.streams.StreamsConfig;
@@ -31,6 +32,7 @@
 import org.springframework.kafka.annotation.EnableKafka;
 import org.springframework.kafka.annotation.EnableKafkaStreams;
 import org.springframework.kafka.annotation.KafkaStreamsDefaultConfiguration;
+import org.springframework.kafka.config.KafkaStreamsConfiguration;
 import org.springframework.kafka.test.EmbeddedKafkaBroker;
 import org.springframework.kafka.test.context.EmbeddedKafka;
 import org.springframework.test.annotation.DirtiesContext;
@@ -64,24 +66,25 @@ public void testKafkaStreamsCustomizer() {
 	@Configuration
 	@EnableKafka
 	@EnableKafkaStreams
-	public static class KafkaStreamsConfiguration {
+	public static class KafkaStreamsConfig {
 
 		@Value("${" + EmbeddedKafkaBroker.SPRING_EMBEDDED_KAFKA_BROKERS + "}")
 		private String brokerAddresses;
 
 		@Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_BUILDER_BEAN_NAME)
 		public StreamsBuilderFactoryBean defaultKafkaStreamsBuilder() {
-			StreamsBuilderFactoryBean streamsBuilderFactoryBean = new StreamsBuilderFactoryBean(kStreamsConfigs());
+			StreamsBuilderFactoryBean streamsBuilderFactoryBean =
+					new StreamsBuilderFactoryBean(kStreamsConfigs());
 			streamsBuilderFactoryBean.setKafkaStreamsCustomizer(customizer());
 			return streamsBuilderFactoryBean;
 		}
 
 		@Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
-		public Properties kStreamsConfigs() {
-			Properties props = new Properties();
+		public KafkaStreamsConfiguration kStreamsConfigs() {
+			Map<String, Object> props = new HashMap<>();
 			props.put(StreamsConfig.APPLICATION_ID_CONFIG, APPLICATION_ID);
 			props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, this.brokerAddresses);
-			return props;
+			return new KafkaStreamsConfiguration(props);
 		}
 
 

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

@@ -22,7 +22,8 @@
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.Paths;
-import java.util.Properties;
+import java.util.HashMap;
+import java.util.Map;
 
 import org.apache.kafka.streams.StreamsConfig;
 import org.junit.jupiter.api.BeforeAll;
@@ -36,6 +37,7 @@
 import org.springframework.context.annotation.Configuration;
 import org.springframework.kafka.annotation.EnableKafkaStreams;
 import org.springframework.kafka.annotation.KafkaStreamsDefaultConfiguration;
+import org.springframework.kafka.config.KafkaStreamsConfiguration;
 import org.springframework.kafka.test.EmbeddedKafkaBroker;
 import org.springframework.kafka.test.context.EmbeddedKafka;
 import org.springframework.test.annotation.DirtiesContext;
@@ -79,7 +81,7 @@ public void testCleanupStreams() throws IOException {
 
 	@Configuration
 	@EnableKafkaStreams
-	public static class KafkaStreamsConfiguration {
+	public static class KafkaStreamsConfig {
 
 		@Value("${" + EmbeddedKafkaBroker.SPRING_EMBEDDED_KAFKA_BROKERS + "}")
 		private String brokerAddresses;
@@ -90,12 +92,12 @@ public StreamsBuilderFactoryBean defaultKafkaStreamsBuilder() {
 		}
 
 		@Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
-		public Properties kStreamsConfigs() {
-			Properties props = new Properties();
+		public KafkaStreamsConfiguration kStreamsConfigs() {
+			Map<String, Object> props = new HashMap<>();
 			props.put(StreamsConfig.APPLICATION_ID_CONFIG, APPLICATION_ID);
 			props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, this.brokerAddresses);
 			props.put(StreamsConfig.STATE_DIR_CONFIG, stateStoreDir.toString());
-			return props;
+			return new KafkaStreamsConfiguration(props);
 		}
 
 	}

spring-kafka/src/test/java/org/springframework/kafka/kstream/KafkaStreamsBranchTests.java

@@ -19,9 +19,9 @@
 import static org.assertj.core.api.Assertions.assertThat;
 
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.Properties;
 import java.util.UUID;
 
 import org.apache.kafka.clients.consumer.Consumer;
@@ -43,6 +43,7 @@
 import org.springframework.context.annotation.Configuration;
 import org.springframework.kafka.annotation.EnableKafkaStreams;
 import org.springframework.kafka.annotation.KafkaStreamsDefaultConfiguration;
+import org.springframework.kafka.config.KafkaStreamsConfiguration;
 import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
 import org.springframework.kafka.core.DefaultKafkaProducerFactory;
 import org.springframework.kafka.core.KafkaTemplate;
@@ -56,7 +57,6 @@
 /**
  * @author Elliot Kennedy
  * @author Artem Bilan
- * @author Gary Russell
  *
  * @since 1.3.3
  */
@@ -144,13 +144,13 @@ public Map<String, Object> producerConfigs() {
 		}
 
 		@Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
-		public Properties kStreamsConfigs() {
-			Properties props = new Properties();
+		public KafkaStreamsConfiguration kStreamsConfigs() {
+			Map<String, Object> props = new HashMap<>();
 			props.put(StreamsConfig.APPLICATION_ID_CONFIG, "testStreams");
 			props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, this.brokerAddresses);
 			props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
 			props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
-			return props;
+			return new KafkaStreamsConfiguration(props);
 		}
 
 		@Bean

spring-kafka/src/test/java/org/springframework/kafka/kstream/KafkaStreamsJsonSerializationTests.java

@@ -18,8 +18,8 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 
+import java.util.HashMap;
 import java.util.Map;
-import java.util.Properties;
 import java.util.UUID;
 
 import org.apache.kafka.clients.consumer.Consumer;
@@ -45,6 +45,7 @@
 import org.springframework.context.annotation.Configuration;
 import org.springframework.kafka.annotation.EnableKafkaStreams;
 import org.springframework.kafka.annotation.KafkaStreamsDefaultConfiguration;
+import org.springframework.kafka.config.KafkaStreamsConfiguration;
 import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
 import org.springframework.kafka.core.DefaultKafkaProducerFactory;
 import org.springframework.kafka.core.KafkaTemplate;
@@ -203,11 +204,11 @@ public Map<String, Object> producerConfigs() {
 		}
 
 		@Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
-		public Properties kStreamsConfigs() {
-			Properties props = new Properties();
+		public KafkaStreamsConfiguration kStreamsConfigs() {
+			Map<String, Object> props = new HashMap<>();
 			props.put(StreamsConfig.APPLICATION_ID_CONFIG, "testStreams");
 			props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, this.brokerAddresses);
-			return props;
+			return new KafkaStreamsConfiguration(props);
 		}
 
 		@Bean

spring-kafka/src/test/java/org/springframework/kafka/kstream/KafkaStreamsTests.java

@@ -19,8 +19,8 @@
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.Mockito.mock;
 
+import java.util.HashMap;
 import java.util.Map;
-import java.util.Properties;
 import java.util.UUID;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
@@ -53,6 +53,7 @@
 import org.springframework.kafka.annotation.KafkaStreamsDefaultConfiguration;
 import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
 import org.springframework.kafka.config.KafkaListenerContainerFactory;
+import org.springframework.kafka.config.KafkaStreamsConfiguration;
 import org.springframework.kafka.core.ConsumerFactory;
 import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
 import org.springframework.kafka.core.DefaultKafkaProducerFactory;
@@ -153,7 +154,7 @@ public void testKStreams() throws Exception {
 	@Configuration
 	@EnableKafka
 	@EnableKafkaStreams
-	public static class KafkaStreamsConfiguration {
+	public static class KafkaStreamsConfig {
 
 		@Value("${" + EmbeddedKafkaBroker.SPRING_EMBEDDED_KAFKA_BROKERS + "}")
 		private String brokerAddresses;
@@ -179,16 +180,16 @@ public KafkaTemplate<Integer, String> template() {
 		}
 
 		@Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
-		public Properties kStreamsConfigs() {
-			Properties props = new Properties();
+		public KafkaStreamsConfiguration kStreamsConfigs() {
+			Map<String, Object> props = new HashMap<>();
 			props.put(StreamsConfig.APPLICATION_ID_CONFIG, "testStreams");
 			props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, this.brokerAddresses);
 			props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.Integer().getClass().getName());
 			props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
 			props.put(StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG,
 					WallclockTimestampExtractor.class.getName());
 			props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, "100");
-			return props;
+			return new KafkaStreamsConfiguration(props);
 		}
 
 		@Bean