Spring/spring-kafka
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Remove StreamsConfig dependency

Browse Source 
Remove StreamsConfig dependency from StreamsBuilderFactoryBean
This commit is contained in:
Dharmesh Jogadia
2019-08-14 18:14:40 +05:30
committed by Gary Russell
parent 90e90525bc
commit a34f48d554
2 changed files with 0 additions and 96 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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);

16
spring-kafka/src/test/java/org/springframework/kafka/config/StreamsBuilderFactoryBeanTests.java
View File

@@ -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 {