SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Producer config customization and DLQ issues

Browse Source 
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
This commit is contained in:
Soby Chacko
2022-06-01 16:09:52 -04:00
parent eebf6e2bd2
commit c670efdc08
2 changed files with 26 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1190,10 +1190,11 @@ public class KafkaMessageChannelBinder extends
.getDlqProducerProperties();
KafkaAwareTransactionManager<byte[], byte[]> transMan = transactionManager(
properties.getExtension().getTransactionManager());
final ExtendedProducerProperties<KafkaProducerProperties> 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);

24
binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaRetryDlqBinderOrContainerTests.java
View File

@@ -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();
}
};
}
}
}