From c670efdc08deb0d25eae392bdaccb03bf2e16e2f Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Wed, 1 Jun 2022 16:09:52 -0400 Subject: [PATCH] Producer config customization and DLQ issues When DLQ is enabled on a consumer binding, the internal producer properties used by the DLQ mechanism to send to the topic is not populated with the correct binding name. In the previous version, it was working fine since we were relying on a ThreadLocal. This is revamped recently to properly introduce a binding name as a top level property in ProducerProperties. However, this was not set on the internal producer properties used by the DLQ. This PR addresses this issue. Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/2417 Resolves #2418 --- .../kafka/KafkaMessageChannelBinder.java | 5 ++-- .../KafkaRetryDlqBinderOrContainerTests.java | 24 ++++++++++++++++++- 2 files changed, 26 insertions(+), 3 deletions(-) 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 fe0d365a7..fde9d2d6e 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 @@ -1190,10 +1190,11 @@ public class KafkaMessageChannelBinder extends .getDlqProducerProperties(); KafkaAwareTransactionManager transMan = transactionManager( properties.getExtension().getTransactionManager()); + final ExtendedProducerProperties producerProperties = new ExtendedProducerProperties<>(dlqProducerProperties); + producerProperties.populateBindingName(properties.getBindingName()); ProducerFactory producerFactory = transMan != null ? transMan.getProducerFactory() - : getProducerFactory(null, - new ExtendedProducerProperties<>(dlqProducerProperties), + : getProducerFactory(null, producerProperties, destination.getName() + ".dlq.producer", destination.getName()); final KafkaTemplate kafkaTemplate = new KafkaTemplate<>( producerFactory); diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaRetryDlqBinderOrContainerTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaRetryDlqBinderOrContainerTests.java index 2002ee773..5dc0dbce0 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaRetryDlqBinderOrContainerTests.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaRetryDlqBinderOrContainerTests.java @@ -16,6 +16,8 @@ package org.springframework.cloud.stream.binder.kafka.integration; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; import java.util.function.BiFunction; import java.util.function.Consumer; @@ -28,6 +30,7 @@ import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.cloud.stream.binder.Binding; import org.springframework.cloud.stream.binder.kafka.ListenerContainerWithDlqAndRetryCustomizer; +import org.springframework.cloud.stream.binder.kafka.config.ProducerConfigCustomizer; import org.springframework.cloud.stream.binding.BindingsLifecycleController; import org.springframework.context.annotation.Bean; import org.springframework.kafka.core.KafkaOperations; @@ -48,6 +51,7 @@ import static org.mockito.Mockito.mock; /** * @author Gary Russell + * @author Soby Chacko */ @SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE, properties = { "spring.cloud.function.definition=retryInBinder;retryInContainer", @@ -59,8 +63,10 @@ import static org.mockito.Mockito.mock; @DirtiesContext public class KafkaRetryDlqBinderOrContainerTests { + private static final CountDownLatch latch = new CountDownLatch(2); + @Test - public void retryAndDlqInRightPlace(@Autowired BindingsLifecycleController controller) { + void retryAndDlqInRightPlace(@Autowired BindingsLifecycleController controller) throws Exception { Binding retryInBinder = controller.queryState("retryInBinder-in-0"); assertThat(KafkaTestUtils.getPropertyValue(retryInBinder, "lifecycle.retryTemplate")).isNotNull(); assertThat(KafkaTestUtils.getPropertyValue(retryInBinder, @@ -72,6 +78,9 @@ public class KafkaRetryDlqBinderOrContainerTests { assertThat(KafkaTestUtils.getPropertyValue(retryInContainer, "lifecycle.messageListenerContainer.commonErrorHandler.failureTracker.backOff")) .isInstanceOf(ExponentialBackOffWithMaxRetries.class); + boolean await = latch.await(5, TimeUnit.SECONDS); + assertThat(await).isTrue(); + } @SpringBootApplication @@ -112,6 +121,19 @@ public class KafkaRetryDlqBinderOrContainerTests { }; } + // Because we have DLQ enabled on the consumer binding, + // this ProducerConfigCustomizer is used by the producer on DLQ. + @Bean + public ProducerConfigCustomizer producerConfigCustomizer() { + return (producerProperties, binding, destination) -> { + if (binding.equals("retryInBinder-in-0")) { + latch.countDown(); + } + else if (binding.equals("retryInContainer-in-0")) { + latch.countDown(); + } + }; + } } }