Binding name as a consumer/producer property

Make binding names available through Consumer/Producer properties.
Currently, the binders use a ThreadLocal to store the binding name for internal use.
These changes introduce the binding name as a property for both producer/consumer bindings.

Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/2380

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2021 the original author or authors.
+ * Copyright 2014-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.
@@ -207,8 +207,6 @@ public class KafkaMessageChannelBinder extends
 	 */
 	public static final String X_ORIGINAL_TIMESTAMP_TYPE = "x-original-timestamp-type";
 
-	private static final ThreadLocal<String> bindingNameHolder = new ThreadLocal<>();
-
 	private static final Pattern interceptorNeededPattern = Pattern.compile("(payload|#root|#this)");
 
 	private static final SpelExpressionParser PARSER = new SpelExpressionParser();
@@ -337,13 +335,11 @@ public class KafkaMessageChannelBinder extends
 
 	@Override
 	public KafkaConsumerProperties getExtendedConsumerProperties(String channelName) {
-		bindingNameHolder.set(channelName);
 		return this.extendedBindingProperties.getExtendedConsumerProperties(channelName);
 	}
 
 	@Override
 	public KafkaProducerProperties getExtendedProducerProperties(String channelName) {
-		bindingNameHolder.set(channelName);
 		return this.extendedBindingProperties.getExtendedProducerProperties(channelName);
 	}
 
@@ -573,8 +569,7 @@ public class KafkaMessageChannelBinder extends
 			props.putAll(kafkaProducerProperties.getConfiguration());
 		}
 		if (this.producerConfigCustomizer != null) {
-			this.producerConfigCustomizer.configure(props, bindingNameHolder.get(), destination);
-			bindingNameHolder.remove();
+			this.producerConfigCustomizer.configure(props, producerProperties.getBindingName(), destination);
 		}
 		DefaultKafkaProducerFactory<byte[], byte[]> producerFactory = new DefaultKafkaProducerFactory<>(
 				props);
@@ -855,8 +850,7 @@ public class KafkaMessageChannelBinder extends
 
 		Assert.isTrue(!extendedConsumerProperties.getExtension().isResetOffsets(),
 				"'resetOffsets' cannot be set when a KafkaBindingRebalanceListener is provided");
-		final String bindingName = bindingNameHolder.get();
-		bindingNameHolder.remove();
+		final String bindingName = extendedConsumerProperties.getBindingName();
 		Assert.notNull(bindingName, "'bindingName' cannot be null");
 		final KafkaBindingRebalanceListener userRebalanceListener = this.rebalanceListener;
 		containerProperties
@@ -1457,7 +1451,7 @@ public class KafkaMessageChannelBinder extends
 		}
 
 		if (this.consumerConfigCustomizer != null) {
-			this.consumerConfigCustomizer.configure(props, bindingNameHolder.get(), destination);
+			this.consumerConfigCustomizer.configure(props, consumerProperties.getBindingName(), destination);
 		}
 		DefaultKafkaConsumerFactory<Object, Object> factory = new DefaultKafkaConsumerFactory<>(props);
 		factory.setBeanName(beanName);

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

@@ -1157,7 +1157,7 @@ public class KafkaBinderTests extends
 			assertThat(receivedMessage.getHeaders()
 					.get(KafkaMessageChannelBinder.X_EXCEPTION_FQCN)).isNotNull();
 			assertThat(receivedMessage.getHeaders()
-					.get(KafkaHeaders.RECEIVED_KEY)).isEqualTo(expectedDlqPartition);
+					.get(KafkaHeaders.RECEIVED_PARTITION)).isEqualTo(expectedDlqPartition);
 		}
 		else if (!HeaderMode.none.equals(headerMode)) {
 			assertThat(handler.getInvocationCount())
@@ -1205,7 +1205,7 @@ public class KafkaBinderTests extends
 					.get(KafkaMessageChannelBinder.X_EXCEPTION_FQCN)).isNotNull();
 
 			assertThat(receivedMessage.getHeaders()
-					.get(KafkaHeaders.RECEIVED_KEY)).isEqualTo(expectedDlqPartition);
+					.get(KafkaHeaders.RECEIVED_PARTITION)).isEqualTo(expectedDlqPartition);
 		}
 		else {
 			assertThat(receivedMessage.getHeaders()

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-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.
@@ -39,6 +39,11 @@ import org.springframework.messaging.Message;
 @JsonInclude(JsonInclude.Include.NON_DEFAULT)
 public class ConsumerProperties {
 
+	/**
+	 * Binding name for this consumer binding.
+	 */
+	private String bindingName;
+
 	/**
 	 * Signals if this consumer needs to be started automatically.
 	 *
@@ -175,6 +180,19 @@ public class ConsumerProperties {
 	 */
 	private boolean batchMode;
 
+	public String getBindingName() {
+		return bindingName;
+	}
+
+	/**
+	 * This method is not intended as a configuration property to set by the applications.
+	 * Therefore, we are not providing a proper setter method for this.
+	 * @param bindingName binding name populated by the framework.
+	 */
+	public void populateBindingName(String bindingName) {
+		this.bindingName = bindingName;
+	}
+
 	public String getRetryTemplateName() {
 		return retryTemplateName;
 	}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-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.
@@ -40,6 +40,11 @@ import org.springframework.expression.Expression;
 @JsonInclude(Include.NON_DEFAULT)
 public class ProducerProperties {
 
+	/**
+	 * Binding name for this producer binding.
+	 */
+	private String bindingName;
+
 	/**
 	 * Signals if this producer needs to be started automatically. Default: true
 	 */
@@ -77,6 +82,19 @@ public class ProducerProperties {
 
 	private PollerProperties poller;
 
+	public String getBindingName() {
+		return bindingName;
+	}
+
+	/**
+	 * This method is not intended as a configuration property to set by the applications.
+	 * Therefore, we are not providing a proper setter method for this.
+	 * @param bindingName binding name populated by the framework.
+	 */
+	public void populateBindingName(String bindingName) {
+		this.bindingName = bindingName;
+	}
+
 	public Expression getPartitionKeyExpression() {
 		return this.partitionKeyExpression;
 	}

core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingService.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2021 the original author or authors.
+ * Copyright 2015-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.
@@ -116,7 +116,7 @@ public class BindingService {
 
 			consumerProperties = extendedConsumerProperties;
 		}
-
+		consumerProperties.populateBindingName(inputName);
 		validate(consumerProperties);
 
 		String bindingTarget = this.bindingServiceProperties
@@ -288,6 +288,7 @@ public class BindingService {
 
 			producerProperties = extendedProducerProperties;
 		}
+		producerProperties.populateBindingName(outputName);
 		validate(producerProperties);
 		Binding<T> binding = doBindProducer(output, bindingTarget, binder,
 				producerProperties);

core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/BindingServiceTests.java

@@ -27,6 +27,7 @@ import java.util.Map;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
+import java.util.function.Function;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 import org.junit.jupiter.api.Test;
@@ -538,6 +539,20 @@ public class BindingServiceTests {
 		assertThat(inputBinding.isRunning()).isFalse();
 	}
 
+	@SuppressWarnings("unchecked")
+	@Test
+	void testBindingNameAsTopLevelProperty() throws Exception {
+		ApplicationContext context = new SpringApplicationBuilder(BarConfiguration.class)
+			.web(WebApplicationType.NONE).run();
+
+		final BindingServiceProperties bindingServiceProperties = context.getBean(BindingServiceProperties.class);
+
+		final ConsumerProperties consumerProperties = bindingServiceProperties.getConsumerProperties("myFunction-in-0");
+		assertThat(consumerProperties.getBindingName()).isEqualTo("myFunction-in-0");
+		final ProducerProperties producerProperties = bindingServiceProperties.getProducerProperties("myFunction-out-0");
+		assertThat(producerProperties.getBindingName()).isEqualTo("myFunction-out-0");
+	}
+
 	private DefaultBinderFactory createMockBinderFactory() {
 		BinderTypeRegistry binderTypeRegistry = createMockBinderTypeRegistry();
 		return new DefaultBinderFactory(
@@ -580,6 +595,17 @@ public class BindingServiceTests {
 
 	}
 
+	@Import(TestChannelBinderConfiguration.class)
+	@EnableAutoConfiguration
+	public static class BarConfiguration {
+
+		@Bean
+		public Function<String, String> myFunction() {
+			return s -> s;
+		}
+
+	}
+
 	public static class FooBinder
 			implements Binder<SomeBindableType, ConsumerProperties, ProducerProperties> {