Improve KafkaStreams ineligible-definitions property handling

* KafkaStreamsBinderEnvironmentPostProcessor respects existing ineligible-definitions
* Add tests for KafkaStreamsBinderEnvironmentPostProcessor

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<String, Object> 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)));
 	}
 
 }

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<KStream<Object, String>> logInput() {
+			return s -> {
+				// No-op consumer
+			};
+		}
+	}
+}