diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java index e0ff6c06a..e4dbc06cb 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java +++ b/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 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 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 factory = new DefaultKafkaConsumerFactory<>(props); factory.setBeanName(beanName); diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java index 83a5507e5..5db4c93a2 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java +++ b/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() diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ConsumerProperties.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ConsumerProperties.java index 4da08791f..afac37e47 100644 --- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ConsumerProperties.java +++ b/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; } diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ProducerProperties.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ProducerProperties.java index 996a61ba7..716f58d6e 100644 --- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ProducerProperties.java +++ b/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; } diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingService.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingService.java index 81e2e4a24..938874841 100644 --- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingService.java +++ b/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 binding = doBindProducer(output, bindingTarget, binder, producerProperties); diff --git a/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/BindingServiceTests.java b/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/BindingServiceTests.java index 373c3082f..115d5eb59 100644 --- a/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/BindingServiceTests.java +++ b/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 myFunction() { + return s -> s; + } + + } + public static class FooBinder implements Binder {