From 6dd67742800fd67586754040d5305132d06fc7a8 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 beafadb6a..db20957c9 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 @@ -1094,10 +1094,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 95d6d37fa..f5aec50f0 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.support.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 - 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(); + } + }; + } } }