From a34f48d5540f3f8007f3d4a6e7b79f10572acf6e Mon Sep 17 00:00:00 2001
From: Dharmesh Jogadia <dharmeshjogadia@gmail.com>
Date: Wed, 14 Aug 2019 18:14:40 +0530
Subject: [PATCH] Remove StreamsConfig dependency

Remove StreamsConfig dependency from StreamsBuilderFactoryBean
---
 .../config/StreamsBuilderFactoryBean.java     | 80 -------------------
 .../StreamsBuilderFactoryBeanTests.java       | 16 ----
 2 files changed, 96 deletions(-)

diff --git a/spring-kafka/src/main/java/org/springframework/kafka/config/StreamsBuilderFactoryBean.java b/spring-kafka/src/main/java/org/springframework/kafka/config/StreamsBuilderFactoryBean.java
index 34387222..9e38c09e 100644
--- a/spring-kafka/src/main/java/org/springframework/kafka/config/StreamsBuilderFactoryBean.java
+++ b/spring-kafka/src/main/java/org/springframework/kafka/config/StreamsBuilderFactoryBean.java
@@ -24,7 +24,6 @@ import org.apache.commons.logging.LogFactory;
 import org.apache.kafka.streams.KafkaClientSupplier;
 import org.apache.kafka.streams.KafkaStreams;
 import org.apache.kafka.streams.StreamsBuilder;
-import org.apache.kafka.streams.StreamsConfig;
 import org.apache.kafka.streams.Topology;
 import org.apache.kafka.streams.processor.StateRestoreListener;
 import org.apache.kafka.streams.processor.internals.DefaultKafkaClientSupplier;
@@ -103,32 +102,6 @@ public class StreamsBuilderFactoryBean extends AbstractFactoryBean<StreamsBuilde
 		this.cleanupConfig = new CleanupConfig();
 	}
 
-	/**
-	 * 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());
-	}
-
-	/**
-	 * Construct an instance with the supplied streams configuration and
-	 * clean up configuration.
-	 * @param streamsConfig the streams configuration.
-	 * @param cleanupConfig the cleanup configuration.
-	 * @since 2.1.2.
-	 * @deprecated in favor of {@link #StreamsBuilderFactoryBean(KafkaStreamsConfiguration, CleanupConfig)}
-	 */
-	@Deprecated
-	public StreamsBuilderFactoryBean(StreamsConfig streamsConfig, CleanupConfig cleanupConfig) {
-		Assert.notNull(streamsConfig, STREAMS_CONFIG_MUST_NOT_BE_NULL);
-		Assert.notNull(cleanupConfig, CLEANUP_CONFIG_MUST_NOT_BE_NULL);
-		this.properties = propertiesFromStreamsConfig(streamsConfig);
-		this.cleanupConfig = cleanupConfig;
-	}
-
 	/**
 	 * Construct an instance with the supplied streams configuration and
 	 * clean up configuration.
@@ -143,16 +116,6 @@ public class StreamsBuilderFactoryBean extends AbstractFactoryBean<StreamsBuilde
 		this.cleanupConfig = cleanupConfig;
 	}
 
-	/**
-	 * Construct an instance with the supplied streams configuration.
-	 * @param streamsConfig the streams configuration.
-	 * @deprecated in favor of {@link #StreamsBuilderFactoryBean(KafkaStreamsConfiguration)}.
-	 */
-	@Deprecated
-	public StreamsBuilderFactoryBean(Map<String, Object> streamsConfig) {
-		this(streamsConfig, new CleanupConfig());
-	}
-
 	/**
 	 * Construct an instance with the supplied streams configuration.
 	 * @param streamsConfig the streams configuration.
@@ -162,45 +125,6 @@ public class StreamsBuilderFactoryBean extends AbstractFactoryBean<StreamsBuilde
 		this(streamsConfig, new CleanupConfig());
 	}
 
-	/**
-	 * Construct an instance with the supplied streams configuration and
-	 * clean up configuration.
-	 * @param streamsConfig the streams configuration.
-	 * @param cleanupConfig the cleanup configuration.
-	 * @since 2.1.2.
-	 * @deprecated in favor of {@link #StreamsBuilderFactoryBean(KafkaStreamsConfiguration, CleanupConfig)}.
-	 */
-	@Deprecated
-	public StreamsBuilderFactoryBean(Map<String, Object> streamsConfig, CleanupConfig cleanupConfig) {
-		Assert.notNull(streamsConfig, STREAMS_CONFIG_MUST_NOT_BE_NULL);
-		Assert.notNull(cleanupConfig, CLEANUP_CONFIG_MUST_NOT_BE_NULL);
-		this.properties = propertiesFromConfigs(streamsConfig);
-		this.cleanupConfig = cleanupConfig;
-	}
-
-	/**
-	 * Set {@link StreamsConfig} on this factory.
-	 * @param streamsConfig the streams configuration.
-	 * @deprecated in favor of {@link #setStreamsConfiguration(Properties)}.
-	 * @since 2.1.3
-	 */
-	@Deprecated
-	public void setStreamsConfig(StreamsConfig streamsConfig) {
-		Assert.notNull(streamsConfig, STREAMS_CONFIG_MUST_NOT_BE_NULL);
-		Assert.isNull(this.properties, "Cannot have both streamsConfig and streams configuration properties");
-		this.properties = propertiesFromStreamsConfig(streamsConfig);
-	}
-
-	/**
-	 * Get the streams config.
-	 * @return the config.
-	 * @deprecated in favor of {@link #getStreamsConfiguration()}.
-	 */
-	@Deprecated
-	public StreamsConfig getStreamsConfig() {
-		return new StreamsConfig(this.properties);
-	}
-
 	/**
 	 * Set {@link StreamsConfig} on this factory.
 	 * @param streamsConfig the streams configuration.
@@ -357,10 +281,6 @@ public class StreamsBuilderFactoryBean extends AbstractFactoryBean<StreamsBuilde
 		return this.running;
 	}
 
-	private Properties propertiesFromStreamsConfig(StreamsConfig config) {
-		return propertiesFromConfigs(config.originals());
-	}
-
 	private Properties propertiesFromConfigs(Map<String, Object> configs) {
 		Properties props = new Properties();
 		props.putAll(configs);
diff --git a/spring-kafka/src/test/java/org/springframework/kafka/config/StreamsBuilderFactoryBeanTests.java b/spring-kafka/src/test/java/org/springframework/kafka/config/StreamsBuilderFactoryBeanTests.java
index f1b537b6..d1d75d5d 100644
--- a/spring-kafka/src/test/java/org/springframework/kafka/config/StreamsBuilderFactoryBeanTests.java
+++ b/spring-kafka/src/test/java/org/springframework/kafka/config/StreamsBuilderFactoryBeanTests.java
@@ -101,22 +101,6 @@ public class StreamsBuilderFactoryBeanTests {
 		verify(streamsBuilder).build(kafkaStreamsConfiguration.asProperties());
 	}
 
-	@Test
-	@SuppressWarnings("deprecation")
-	public void testBuildWithStreamsConfig() throws Exception {
-		StreamsConfig streamsConfig = new StreamsConfig(kafkaStreamsConfiguration.asProperties());
-		streamsBuilderFactoryBean = new StreamsBuilderFactoryBean(streamsConfig) {
-			@Override
-			protected StreamsBuilder createInstance() {
-				return spy(super.createInstance());
-			}
-		};
-		streamsBuilderFactoryBean.afterPropertiesSet();
-		streamsBuilderFactoryBean.start();
-		StreamsBuilder streamsBuilder = streamsBuilderFactoryBean.getObject();
-		verify(streamsBuilder).build(kafkaStreamsConfiguration.asProperties());
-	}
-
 	@Configuration
 	@EnableKafkaStreams
 	public static class KafkaStreamsConfig {