From 200a5efb640f4a03e4549c0486b87835fde45b56 Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Fri, 9 Oct 2020 20:32:46 -0400 Subject: [PATCH] Simplifying bootstrap-server config evaluation 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 --- .../KafkaBinderConfigurationProperties.java | 19 ++++--------------- 1 file changed, 4 insertions(+), 15 deletions(-) diff --git a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java index 199a085fd..0dab3deeb 100644 --- a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java +++ b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java @@ -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 getConfigurationWithBootstrapServer( Map 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 bootStrapServers = (List) 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); }