GH-3124 Add testing for accessing additional configuration properties

Author: olegz

binders/kafka-binder/spring-cloud-stream-binder-kafka/pom.xml

@@ -46,6 +46,10 @@
 			<groupId>org.springframework.kafka</groupId>
 			<artifactId>spring-kafka</artifactId>
 		</dependency>
+		<dependency>
+			<groupId>com.fasterxml.jackson.datatype</groupId>
+			<artifactId>jackson-datatype-jsr310</artifactId>
+		</dependency>
 		<dependency>
 			<groupId>org.springframework.boot</groupId>
 			<artifactId>spring-boot-test</artifactId>

binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java

@@ -40,6 +40,8 @@
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializationFeature;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
@@ -250,6 +252,9 @@ public class KafkaMessageChannelBinder extends
 
 	private final KafkaAdmin kafkaAdmin;
 
+	// used strictly for serializeing additional configuration properties. See 'doGetAdditionalConfigurationProperties'
+	private ObjectMapper objectMapper = new ObjectMapper();
+
 	public KafkaMessageChannelBinder(
 			KafkaBinderConfigurationProperties configurationProperties,
 			KafkaTopicProvisioner provisioningProvider) {
@@ -284,6 +289,8 @@ public KafkaMessageChannelBinder(
 		this.dlqPartitionFunction = dlqPartitionFunction;
 		this.dlqDestinationResolver = dlqDestinationResolver;
 		this.kafkaAdmin = new KafkaAdmin(new HashMap<>(provisioningProvider.getAdminClientProperties()));
+		this.objectMapper.findAndRegisterModules();
+		this.objectMapper.configure(SerializationFeature.FAIL_ON_EMPTY_BEANS, false);
 	}
 
 	@Override
@@ -851,14 +858,30 @@ private void propagateContainerProperties(ContainerProperties containerPropertie
 		containerProperties.setStopImmediate(listener.isImmediateStop());
 	}
 
+	/**
+	 * Returns an unmodifiable copy of {@link ContainerProperties} associated with the destination name
+	 * which corresponds to a particular binding which could be accessed under 'containerProperties' key.
+	 *
+	 * @param destinationName the name of the destination (or binding name if destination is not specified)
+	 * @return map of
+	 */
+	@Override
+	protected Map<String, Object> doGetAdditionalConfigurationProperties(String destinationName) {
+		ContainerProperties kafkaContainerProperties = this.kafkaMessageListenerContainers.iterator().next().getContainerProperties();
+		Map mapOfContainerProperties = this.objectMapper.convertValue(kafkaContainerProperties, Map.class);
+		Map<String, Object> additionalConfigurationProperties = new HashMap<>();
+		additionalConfigurationProperties.put("containerProperties", mapOfContainerProperties);
+		return additionalConfigurationProperties;
+	}
+
 	private BiFunction<ConsumerRecord<?, ?>, Exception, TopicPartition> createDestResolver(
-			KafkaConsumerProperties extension) {
+		KafkaConsumerProperties extension) {
 
 		Integer dlqPartitions = extension.getDlqPartitions();
 		if (extension.isEnableDlq()) {
 			return (rec, ex) -> dlqPartitions == null || dlqPartitions > 1
-						? new TopicPartition(extension.getDlqName(), rec.partition())
-						: new TopicPartition(extension.getDlqName(), 0);
+				? new TopicPartition(extension.getDlqName(), rec.partition())
+				: new TopicPartition(extension.getDlqName(), 0);
 		}
 		else {
 			return null;

binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaConfigurationTests.java

@@ -0,0 +1,78 @@
+/*
+ * Copyright 2024-2025 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;
+
+import java.util.Map;
+import java.util.function.Consumer;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.cloud.stream.binder.Binding;
+import org.springframework.cloud.stream.binding.BindingService;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.test.context.EmbeddedKafka;
+import org.springframework.test.annotation.DirtiesContext;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * @author Oleg Zhurakousky
+ */
+@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE, properties = {
+	"spring.cloud.function.definition=barConsumer;fooConsumer",
+	"spring.kafka.listener.immediate-stop=true",
+	"spring.cloud.stream.bindings.fooConsumer-in-0.destination=foo"
+})
+@EmbeddedKafka
+@DirtiesContext
+public class KafkaConfigurationTests {
+
+
+	@Autowired
+	private BindingService bindingService;
+
+	@Test
+	void testKafkaContainerConfigurationPropagation() throws Exception {
+		Binding<?> fooDestination = this.bindingService.getConsumerBindings("fooConsumer-in-0").iterator().next();
+		Map<String, Object> fooAdditionalConfigurationProperties = fooDestination.getAdditionalConfigurationProperties();
+		assertThat(((Map) fooAdditionalConfigurationProperties.get("containerProperties")).get("stopImmediate")).isEqualTo(true);
+
+		Binding<?> barDestination = this.bindingService.getConsumerBindings("barConsumer-in-0").iterator().next();
+		Map<String, Object> barAdditionalConfigurationProperties = barDestination.getAdditionalConfigurationProperties();
+		assertThat(((Map) barAdditionalConfigurationProperties.get("containerProperties")).get("stopImmediate")).isEqualTo(true);
+	}
+
+	@EnableAutoConfiguration
+	@Configuration
+	public static class Config {
+
+		@Bean
+		Consumer<String> barConsumer() {
+			return message -> {
+			};
+		}
+		@Bean
+		Consumer<String> fooConsumer() {
+			return message -> {
+			};
+		}
+	}
+}

core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java

@@ -582,6 +582,11 @@ public String getBinderType() {
 					return resolveBinderType(getBindingName(), bsp);
 				}
 
+				@Override
+				public Map<String, Object> getAdditionalConfigurationProperties() {
+					return doGetAdditionalConfigurationProperties(this.getName());
+				}
+
 				@Override
 				protected void afterUnbind() {
 					try {
@@ -617,6 +622,17 @@ else if (e instanceof ProvisioningException provisioningException) {
 		}
 	}
 
+	/**
+	 * This method must be implemented by an individual binders to produce an
+	 * immutable version of additional configuration properties primarily for testing and diagnosing/debugging issues.
+	 * @return map pf additional configuration properties as key/vaalue.
+	 */
+	protected Map<String, Object> doGetAdditionalConfigurationProperties(String name) {
+		logger.warn("This method must be implemented by an individual binders to produce " +
+			"an immutable version of additional configuration properties primarily for testing and diagnosing/debugging issues");
+		return null;
+	}
+
 	@Override
 	public Binding<PollableSource<MessageHandler>> bindPollableConsumer(String name,
 			String group, final PollableSource<MessageHandler> inboundBindTarget,

core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/Binding.java

@@ -50,6 +50,16 @@ default <P> P getExtension() {
 		return null;
 	}
 
+	/**
+	 * Will return additional configuration properties associated with this binding.
+	 * Must be implemented by individual binders and is primarily designed for testing and
+	 * debugging/diagnosing issues.
+	 * @return  additional configuration properties associated with this binding.
+	 */
+	default Map<String, Object> getAdditionalConfigurationProperties() {
+		return Collections.emptyMap();
+	}
+
 	/**
 	 * Starts the target component represented by this instance. NOTE: At the time the
 	 * instance is created the component is already started. This operation is typically