From 3fdc4784a328f904d2314dd0397977caf9ed5e7a Mon Sep 17 00:00:00 2001 From: Sudhanshu Ratna Thakur Date: Mon, 19 Oct 2026 22:20:12 +0530 Subject: [PATCH] Support broker lists in Kafka Streams binder fallback Fixes gh-3307 Signed-off-by: Sudhanshu Ratna Thakur --- ...StreamsBinderSupportAutoConfiguration.java | 18 ++++-- ...msBinderSupportAutoConfigurationTests.java | 55 +++++++++++++++++++ .../KafkaStreamsBinderBootstrapTest.java | 17 ++++++ 3 files changed, 85 insertions(+), 5 deletions(-) diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java index 591a1aa1bf..d453e8da08 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java @@ -196,16 +196,24 @@ public Map streamConfigGlobalProperties( String kafkaConnectionString = configProperties.getKafkaConnectionString(); if (kafkaConnectionString != null && kafkaConnectionString.equals("localhost:9092")) { + Binder binder = Binder.get(environment); //Making sure that the application indeed set a property. - String kafkaStreamsBinderBroker = environment.getProperty("spring.cloud.stream.kafka.streams.binder.brokers"); + String[] kafkaStreamsBinderBrokers = binder.bind("spring.cloud.stream.kafka.streams.binder.brokers", + String[].class).orElse(null); + String kafkaStreamsBinderBroker = ObjectUtils.isEmpty(kafkaStreamsBinderBrokers) ? null + : StringUtils.arrayToCommaDelimitedString(kafkaStreamsBinderBrokers); if (!StringUtils.hasText(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.hasText(kafkaBinderBroker)) { - kafkaConnectionString = kafkaBinderBroker; - configProperties.setBrokers(kafkaConnectionString); + String[] kafkaBinderBrokers = binder.bind("spring.cloud.stream.kafka.binder.brokers", + String[].class).orElse(null); + if (!ObjectUtils.isEmpty(kafkaBinderBrokers)) { + String kafkaBinderBroker = StringUtils.arrayToCommaDelimitedString(kafkaBinderBrokers); + if (StringUtils.hasText(kafkaBinderBroker)) { + kafkaConnectionString = kafkaBinderBroker; + configProperties.setBrokers(kafkaBinderBrokers); + } } } } diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfigurationTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfigurationTests.java index 4cbcaff086..403de1f077 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfigurationTests.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfigurationTests.java @@ -16,8 +16,11 @@ package org.springframework.cloud.stream.binder.kafka.streams; +import java.util.Map; import java.util.stream.Stream; +import org.apache.kafka.streams.StreamsConfig; + import org.junit.jupiter.api.Test; import org.springframework.beans.factory.ObjectProvider; @@ -26,6 +29,8 @@ import org.springframework.cloud.stream.config.BindingServiceProperties; import org.springframework.cloud.stream.function.StreamFunctionProperties; import org.springframework.core.env.ConfigurableEnvironment; +import org.springframework.mock.env.MockEnvironment; +import org.springframework.kafka.config.KafkaStreamsConfiguration; import org.springframework.kafka.config.StreamsBuilderFactoryBeanConfigurer; import org.springframework.kafka.core.CleanupConfig; @@ -40,6 +45,56 @@ */ class KafkaStreamsBinderSupportAutoConfigurationTests { + @Test + void kafkaBinderBrokerListIsUsedAsStreamsFallback() { + MockEnvironment environment = new MockEnvironment() + .withProperty("spring.cloud.stream.kafka.binder.brokers[0]", "host1:9092") + .withProperty("spring.cloud.stream.kafka.binder.brokers[1]", "host2:9092"); + + assertThat(streamConfigGlobalProperties(environment).get(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG)) + .isEqualTo("host1:9092,host2:9092"); + } + + @Test + void kafkaBinderBrokerStringIsUsedAsStreamsFallback() { + MockEnvironment environment = new MockEnvironment() + .withProperty("spring.cloud.stream.kafka.binder.brokers", "host1:9092,host2:9092"); + + assertThat(streamConfigGlobalProperties(environment).get(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG)) + .isEqualTo("host1:9092,host2:9092"); + } + + @Test + void streamsBrokerListPreventsKafkaBinderFallback() { + MockEnvironment environment = new MockEnvironment() + .withProperty("spring.cloud.stream.kafka.streams.binder.brokers[0]", "streams-host:9092") + .withProperty("spring.cloud.stream.kafka.binder.brokers[0]", "kafka-host:9092"); + + assertThat(streamConfigGlobalProperties(environment).get(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG)) + .isEqualTo("localhost:9092"); + } + + @Test + void emptyKafkaBinderBrokersLeaveStreamsDefaultUnchanged() { + MockEnvironment environment = new MockEnvironment() + .withProperty("spring.cloud.stream.kafka.binder.brokers", ""); + + assertThat(streamConfigGlobalProperties(environment).get(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG)) + .isEqualTo("localhost:9092"); + } + + private Map streamConfigGlobalProperties(MockEnvironment environment) { + KafkaStreamsBinderConfigurationProperties binderProperties = mock(KafkaStreamsBinderConfigurationProperties.class); + when(binderProperties.getKafkaConnectionString()).thenReturn("localhost:9092"); + when(binderProperties.mergedConsumerConfiguration()).thenReturn(Map.of()); + when(binderProperties.mergedProducerConfiguration()).thenReturn(Map.of()); + + return new KafkaStreamsBinderSupportAutoConfiguration().streamConfigGlobalProperties( + binderProperties, + new KafkaStreamsConfiguration(Map.of(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092")), + environment, mock(SendToDlqAndContinue.class)); + } + @Test @SuppressWarnings("unchecked") void kafkaStreamsFunctionProcessorUsesAllStreamsBuilderFactoryBeanConfigurers() { diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/bootstrap/KafkaStreamsBinderBootstrapTest.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/bootstrap/KafkaStreamsBinderBootstrapTest.java index 687d9386cb..27487b7a38 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/bootstrap/KafkaStreamsBinderBootstrapTest.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/bootstrap/KafkaStreamsBinderBootstrapTest.java @@ -16,6 +16,7 @@ package org.springframework.cloud.stream.binder.kafka.streams.bootstrap; +import java.util.List; import java.util.Map; import java.util.Properties; import java.util.function.Consumer; @@ -24,6 +25,7 @@ import tools.jackson.databind.type.TypeFactory; import org.apache.kafka.common.header.Headers; import org.apache.kafka.common.security.JaasUtils; +import org.apache.kafka.streams.StreamsConfig; import org.apache.kafka.streams.kstream.GlobalKTable; import org.apache.kafka.streams.kstream.KStream; import org.apache.kafka.streams.kstream.KTable; @@ -82,6 +84,21 @@ void kafkaStreamsBinderWithStandardConfigCanStart(boolean excludeKafkaAutoConfig } // @checkstyle:on } + @Test + @SuppressWarnings("unchecked") + void kafkaStreamsBinderUsesKafkaBinderBrokerList() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder(SimpleKafkaStreamsApplication.class) + .web(WebApplicationType.NONE).run( + "--spring.cloud.function.definition=input1", + "--spring.cloud.stream.kafka.streams.bindings.input1-in-0.consumer.application-id" + + "=testKafkaStreamsBinderUsesKafkaBinderBrokerList", + "--spring.cloud.stream.kafka.binder.brokers[0]=" + embeddedKafka.getBrokersAsString())) { + Map properties = context.getBean("streamConfigGlobalProperties", Map.class); + assertThat(properties.get(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG)) + .isEqualTo(List.of(embeddedKafka.getBrokersAsString())); + } + } + @ParameterizedTest @ValueSource(booleans = { false, true }) void kafkaStreamsBinderWithCustomConfigCanStart(boolean excludeKafkaAutoConfig) {