From 6a8c0cd0c6cf2bf9a156ca74d0de9cac538055aa Mon Sep 17 00:00:00 2001 From: Ilayaperumal Gopinathan Date: Tue, 18 Apr 2017 10:49:38 +0530 Subject: [PATCH] Fix KafkaHealthIndicator kafka properties - Override KafkaHealthIndicator's `bootstrap.servers` property only when it is not set already - Add test Resolves #123 --- .../config/KafkaBinderConfiguration.java | 4 +++- ...BinderAutoConfigurationPropertiesTest.java | 22 +++++++++++++++++++ 2 files changed, 25 insertions(+), 1 deletion(-) diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java index c9a28ef98..ffec39078 100644 --- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java @@ -125,7 +125,9 @@ public class KafkaBinderConfiguration { if (!ObjectUtils.isEmpty(configurationProperties.getConfiguration())) { props.putAll(configurationProperties.getConfiguration()); } - props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, this.configurationProperties.getKafkaConnectionString()); + if (!props.containsKey(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG)) { + props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, this.configurationProperties.getKafkaConnectionString()); + } ConsumerFactory consumerFactory = new DefaultKafkaConsumerFactory<>(props); return new KafkaBinderHealthIndicator(kafkaMessageChannelBinder, consumerFactory); } diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderAutoConfigurationPropertiesTest.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderAutoConfigurationPropertiesTest.java index 9ec3b4a53..7d6eeffd8 100644 --- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderAutoConfigurationPropertiesTest.java +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderAutoConfigurationPropertiesTest.java @@ -35,6 +35,7 @@ import org.springframework.cloud.stream.binder.kafka.config.KafkaBinderConfigura import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties; import org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties; import org.springframework.context.annotation.Bean; +import org.springframework.kafka.core.ConsumerFactory; import org.springframework.kafka.core.DefaultKafkaConsumerFactory; import org.springframework.kafka.core.DefaultKafkaProducerFactory; import org.springframework.test.context.TestPropertySource; @@ -55,6 +56,9 @@ public class KafkaBinderAutoConfigurationPropertiesTest { @Autowired private KafkaMessageChannelBinder kafkaMessageChannelBinder; + @Autowired + private KafkaBinderHealthIndicator kafkaBinderHealthIndicator; + @Test public void testKafkaBinderConfigurationWithKafkaProperties() throws Exception { assertNotNull(this.kafkaMessageChannelBinder); @@ -85,6 +89,24 @@ public class KafkaBinderAutoConfigurationPropertiesTest { assertTrue((((List) consumerConfigs.get("bootstrap.servers")).containsAll(bootstrapServers))); } + @Test + public void testKafkaHealthIndicatorProperties() { + assertNotNull(this.kafkaBinderHealthIndicator); + Field consumerFactoryField = ReflectionUtils.findField(KafkaBinderHealthIndicator.class, "consumerFactory", + ConsumerFactory.class); + ReflectionUtils.makeAccessible(consumerFactoryField); + DefaultKafkaConsumerFactory consumerFactory = (DefaultKafkaConsumerFactory) ReflectionUtils.getField( + consumerFactoryField, this.kafkaBinderHealthIndicator); + Field configField = ReflectionUtils.findField(DefaultKafkaConsumerFactory.class, "configs", Map.class); + ReflectionUtils.makeAccessible(configField); + Map configs = (Map) ReflectionUtils.getField(configField, consumerFactory); + assertTrue(configs.containsKey("bootstrap.servers")); + List bootstrapServers = new ArrayList<>(); + bootstrapServers.add("10.98.09.199:9092"); + bootstrapServers.add("10.98.09.196:9092"); + assertTrue(((List)configs.get("bootstrap.servers")).containsAll(bootstrapServers)); + } + public static class KafkaBinderConfigProperties { @Bean