From 46f1b418327b95feb02251ec18489a4f5e49a42b Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Fri, 9 Aug 2019 18:54:13 -0400 Subject: [PATCH] Interop between bootstrap server configuration When using the Kafka Streams binder, if the application chooses to provide bootstrap server configuration through Kafka binder broker property, then allow that. This way either type of broker config works in Kafka Streams binder. Resolves #401 Resolves #720 --- .../GlobalKTableBinderConfiguration.java | 3 +- .../streams/KTableBinderConfiguration.java | 3 +- ...StreamsBinderSupportAutoConfiguration.java | 28 +++++++++++++++---- ...treamsBinderWordCountIntegrationTests.java | 2 +- 4 files changed, 25 insertions(+), 11 deletions(-) diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBinderConfiguration.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBinderConfiguration.java index 2db0a21fe..687971ac4 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBinderConfiguration.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBinderConfiguration.java @@ -23,7 +23,6 @@ import org.springframework.beans.factory.config.BeanFactoryPostProcessor; import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; import org.springframework.boot.autoconfigure.kafka.KafkaProperties; import org.springframework.cloud.stream.annotation.BindingProvider; -import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties; import org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner; import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsBinderConfigurationProperties; import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsExtendedBindingProperties; @@ -48,7 +47,7 @@ public class GlobalKTableBinderConfiguration { @Bean public KafkaTopicProvisioner provisioningProvider( - KafkaBinderConfigurationProperties binderConfigurationProperties, + KafkaStreamsBinderConfigurationProperties binderConfigurationProperties, KafkaProperties kafkaProperties) { return new KafkaTopicProvisioner(binderConfigurationProperties, kafkaProperties); } diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBinderConfiguration.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBinderConfiguration.java index 09423ae63..978e116d8 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBinderConfiguration.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBinderConfiguration.java @@ -23,7 +23,6 @@ import org.springframework.beans.factory.config.BeanFactoryPostProcessor; import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; import org.springframework.boot.autoconfigure.kafka.KafkaProperties; import org.springframework.cloud.stream.annotation.BindingProvider; -import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties; import org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner; import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsBinderConfigurationProperties; import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsExtendedBindingProperties; @@ -48,7 +47,7 @@ public class KTableBinderConfiguration { @Bean public KafkaTopicProvisioner provisioningProvider( - KafkaBinderConfigurationProperties binderConfigurationProperties, + KafkaStreamsBinderConfigurationProperties binderConfigurationProperties, KafkaProperties kafkaProperties) { return new KafkaTopicProvisioner(binderConfigurationProperties, kafkaProperties); } diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java index 79218652f..cbaadb67c 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java @@ -152,14 +152,30 @@ public class KafkaStreamsBinderSupportAutoConfiguration { @Bean("streamConfigGlobalProperties") public Map streamConfigGlobalProperties( KafkaStreamsBinderConfigurationProperties configProperties, - KafkaStreamsConfiguration kafkaStreamsConfiguration) { + KafkaStreamsConfiguration kafkaStreamsConfiguration, ConfigurableEnvironment environment) { Properties properties = kafkaStreamsConfiguration.asProperties(); - // Override Spring Boot bootstrap server setting if left to default with the value - // configured in the binder + + String kafkaConnectionString = configProperties.getKafkaConnectionString(); + + if (kafkaConnectionString != null && kafkaConnectionString.equals("localhost:9092")) { + //Making sure that the application indeed set a property. + String kafkaStreamsBinderBroker = environment.getProperty("spring.cloud.stream.kafka.streams.binder.brokers"); + + if (StringUtils.isEmpty(kafkaStreamsBinderBroker)) { + //Kafka Streams binder specific property for brokers is not set by the application. + //See if there is one configured at the kafka binder level. + String kafkaBinderBroker = environment.getProperty("spring.cloud.stream.kafka.binder.brokers"); + if (!StringUtils.isEmpty(kafkaBinderBroker)) { + kafkaConnectionString = kafkaBinderBroker; + configProperties.setBrokers(kafkaConnectionString); + } + } + } + if (ObjectUtils.isEmpty(properties.get(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG))) { properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, - configProperties.getKafkaConnectionString()); + kafkaConnectionString); } else { Object bootstrapServerConfig = properties @@ -170,14 +186,14 @@ public class KafkaStreamsBinderSupportAutoConfiguration { .get(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG); if (bootStrapServers.equals("localhost:9092")) { properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, - configProperties.getKafkaConnectionString()); + kafkaConnectionString); } } else if (bootstrapServerConfig instanceof List) { List bootStrapCollection = (List) bootstrapServerConfig; if (bootStrapCollection.size() == 1 && bootStrapCollection.get(0).equals("localhost:9092")) { properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, - configProperties.getKafkaConnectionString()); + kafkaConnectionString); } } } diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderWordCountIntegrationTests.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderWordCountIntegrationTests.java index 17b13323f..9cdc93358 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderWordCountIntegrationTests.java +++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderWordCountIntegrationTests.java @@ -116,7 +116,7 @@ public class KafkaStreamsBinderWordCountIntegrationTests { + "=org.apache.kafka.common.serialization.Serdes$StringSerde", "--spring.cloud.stream.kafka.streams.timeWindow.length=5000", "--spring.cloud.stream.kafka.streams.timeWindow.advanceBy=0", - "--spring.cloud.stream.kafka.streams.binder.brokers=" + "--spring.cloud.stream.kafka.binder.brokers=" + embeddedKafka.getBrokersAsString())) { receiveAndValidate(context); }