SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Change Log etc. Replication Factor

Browse Source 
If the user has not explicitly set the `SreamsConfig.REPLICATION_FACTOR_CONFIG`,
set it from the binder property.

This is used for infrastructure topics (change logs and repartition topics).
This commit is contained in:
Gary Russell
2020-10-15 13:26:07 -04:00
committed by Soby Chacko
parent 829d1c651a
commit 50ec8f0919
2 changed files with 8 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java
View File

@@ -247,6 +247,10 @@ public abstract class AbstractKafkaStreamsBinderProcessor implements Application
if (!ObjectUtils.isEmpty(multiBinderKafkaStreamsBinderConfigurationProperties.getConfiguration())) {
streamConfiguration.putAll(multiBinderKafkaStreamsBinderConfigurationProperties.getConfiguration());
}
if (!streamConfiguration.containsKey(StreamsConfig.REPLICATION_FACTOR_CONFIG)) {
streamConfiguration.put(StreamsConfig.REPLICATION_FACTOR_CONFIG,
(int) multiBinderKafkaStreamsBinderConfigurationProperties.getReplicationFactor());
}
}
//this is only used primarily for StreamListener based processors. Although in theory, functions can use it,

4
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java
View File

@@ -270,6 +270,10 @@ public class KafkaStreamsBinderSupportAutoConfiguration {
if (!ObjectUtils.isEmpty(configProperties.getConfiguration())) {
properties.putAll(configProperties.getConfiguration());
}
if (!properties.containsKey(StreamsConfig.REPLICATION_FACTOR_CONFIG)) {
properties.put(StreamsConfig.REPLICATION_FACTOR_CONFIG,
(int) configProperties.getReplicationFactor());
}
return properties.entrySet().stream().collect(
Collectors.toMap((e) -> String.valueOf(e.getKey()), Map.Entry::getValue));
}