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

Simplifying bootstrap-server config evaluation

Browse Source 
If both boot and binder level config for bootstrap servers are present, the boot
one always wins currently regardless of any binder settings, unless the boot one
evaluates to the default (localhost:9092). This is especially a problem in a
multi binder scenario. Addressing this issue by simplifying the evaluation and
always gives the binder config the highest precedence.

Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/967
This commit is contained in:
Soby Chacko
2020-10-09 20:32:46 -04:00
committed by Gary Russell
parent 7f3a7f856f
commit 200a5efb64
1 changed files with 4 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java
View File

@@ -19,7 +19,6 @@ package org.springframework.cloud.stream.binder.kafka.properties;
import java.time.Duration;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import javax.validation.constraints.AssertTrue;
@@ -324,20 +323,10 @@ public class KafkaBinderConfigurationProperties {
private Map<String, Object> getConfigurationWithBootstrapServer(
Map<String, Object> configuration, String bootstrapServersConfig) {
if (ObjectUtils.isEmpty(configuration.get(bootstrapServersConfig))) {
configuration.put(bootstrapServersConfig, getKafkaConnectionString());
}
else {
Object boostrapServersConfig = configuration.get(bootstrapServersConfig);
if (boostrapServersConfig instanceof List) {
@SuppressWarnings("unchecked")
List<String> bootStrapServers = (List<String>) configuration
.get(bootstrapServersConfig);
if (bootStrapServers.size() == 1
&& bootStrapServers.get(0).equals("localhost:9092")) {
configuration.put(bootstrapServersConfig, getKafkaConnectionString());
}
}
final String kafkaConnectionString = getKafkaConnectionString();
if (ObjectUtils.isEmpty(configuration.get(bootstrapServersConfig)) ||
!kafkaConnectionString.equals("localhost:9092")) {
configuration.put(bootstrapServersConfig, kafkaConnectionString);
}
return Collections.unmodifiableMap(configuration);
}