diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderEnvironmentPostProcessor.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderEnvironmentPostProcessor.java index 18a319188..de514f7f0 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderEnvironmentPostProcessor.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderEnvironmentPostProcessor.java @@ -16,8 +16,7 @@ package org.springframework.cloud.stream.binder.kafka.streams; -import java.util.HashMap; -import java.util.Map; +import java.util.Collections; import org.springframework.boot.SpringApplication; import org.springframework.boot.env.EnvironmentPostProcessor; @@ -25,22 +24,30 @@ import org.springframework.core.env.ConfigurableEnvironment; import org.springframework.core.env.MapPropertySource; /** - * {@link EnvironmentPostProcessor} to exclude the SendToDlqAndContinue BiFunction - * in core Spring Cloud Function by adding it to the ineligible function definitions. + * {@link EnvironmentPostProcessor} to ensure the {@link SendToDlqAndContinue sendToDlqAndContinue} BiFunction + * is excluded in core Spring Cloud Function by adding it to the ineligible function definitions. * * @author Soby Chacko + * @author Chris Bono */ public class KafkaStreamsBinderEnvironmentPostProcessor implements EnvironmentPostProcessor { - @Override - public void postProcessEnvironment(ConfigurableEnvironment environment, - SpringApplication application) { - Map kafkaStreamsBinderIneligibleDefns = new HashMap<>(); - kafkaStreamsBinderIneligibleDefns.put("spring.cloud.function.ineligible-definitions", - "sendToDlqAndContinue"); + /** + * The bean name of the SendToDlqAndContinue function - must remain in sync w/ + * {@link KafkaStreamsBinderSupportAutoConfiguration#sendToDlqAndContinue()}. + */ + private static final String SEND_TO_DLQ_AND_CONTINUE_BEAN_NAME = "sendToDlqAndContinue"; - environment.getPropertySources().addLast(new MapPropertySource( - "KAFKA_STREAMS_BINDER_INELIGIBLE_DEFINITIONS", kafkaStreamsBinderIneligibleDefns)); + @Override + public void postProcessEnvironment(ConfigurableEnvironment environment, SpringApplication application) { + String ineligibleDefinitionsPropertyKey = "spring.cloud.function.ineligible-definitions"; + String ineligibleDefinitions = SEND_TO_DLQ_AND_CONTINUE_BEAN_NAME; + if (environment.getProperty(ineligibleDefinitionsPropertyKey) != null) { + ineligibleDefinitions += ("," + environment.getProperty(ineligibleDefinitionsPropertyKey)); + } + environment.getPropertySources().addFirst(new MapPropertySource( + "kafkaStreamsBinderIneligibleDefinitions", + Collections.singletonMap(ineligibleDefinitionsPropertyKey, ineligibleDefinitions))); } } diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderEnvironmentPostProcessorTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderEnvironmentPostProcessorTests.java new file mode 100644 index 000000000..dabdf26b9 --- /dev/null +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderEnvironmentPostProcessorTests.java @@ -0,0 +1,77 @@ +/* + * Copyright 2022-2022 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.binder.kafka.streams; + +import java.util.function.Consumer; + +import org.apache.kafka.streams.kstream.KStream; +import org.junit.jupiter.api.Test; + +import org.springframework.boot.WebApplicationType; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.builder.SpringApplicationBuilder; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.kafka.test.EmbeddedKafkaBroker; +import org.springframework.kafka.test.condition.EmbeddedKafkaCondition; +import org.springframework.kafka.test.context.EmbeddedKafka; + +import static org.assertj.core.api.AssertionsForClassTypes.assertThat; + +/** + * Tests for {@link KafkaStreamsBinderEnvironmentPostProcessor}. + * + * @author Chris Bono + */ +@EmbeddedKafka +public class KafkaStreamsBinderEnvironmentPostProcessorTests { + + private static final EmbeddedKafkaBroker embeddedKafka = EmbeddedKafkaCondition.getBroker(); + + @Test + void defaultIneligibleFunctionIsSet() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder(KafkaStreamsBinderEnvironmentPostProcessorTests.SimpleKafkaStreamsApplication.class) + .web(WebApplicationType.NONE).run( + "--spring.cloud.stream.kafka.streams.binder.brokers=" + + embeddedKafka.getBrokersAsString())) { + assertThat(context.getEnvironment().getProperty("spring.cloud.function.ineligible-definitions")) + .isEqualTo("sendToDlqAndContinue"); + } + } + + @Test + void userSpecifiedIneligibleFunctionIsAppendedToDefaultList() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder(KafkaStreamsBinderEnvironmentPostProcessorTests.SimpleKafkaStreamsApplication.class) + .web(WebApplicationType.NONE).run( + "--spring.cloud.function.ineligible-definitions=foo", + "--spring.cloud.stream.kafka.streams.binder.brokers=" + + embeddedKafka.getBrokersAsString())) { + assertThat(context.getEnvironment().getProperty("spring.cloud.function.ineligible-definitions")) + .isEqualTo("sendToDlqAndContinue,foo"); + } + } + + @EnableAutoConfiguration + static class SimpleKafkaStreamsApplication { + @Bean + public Consumer> logInput() { + return s -> { + // No-op consumer + }; + } + } +}