From 66eb15a8e2d90b1f593f19ddc07869b8d8b1d74c Mon Sep 17 00:00:00 2001 From: Ilayaperumal Gopinathan Date: Fri, 17 Mar 2017 15:31:40 +0530 Subject: [PATCH] Make Kafka DLQ topic name configurable - Make it configurable as a Kafka consumer properties - Add test to verify the configuration - Update doc Resolves #108 Test fix to use the same partition count for producer/dlq Fix KafkaTopicProvisioner in case of configurable dlq topic name - Update test --- .../properties/KafkaConsumerProperties.java | 13 +++- .../provisioning/KafkaTopicProvisioner.java | 4 +- .../src/main/asciidoc/overview.adoc | 7 +- .../kafka/KafkaMessageChannelBinder.java | 9 +-- .../stream/binder/kafka/KafkaBinderTests.java | 64 ++++++++++++++++++- 5 files changed, 89 insertions(+), 8 deletions(-) diff --git a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaConsumerProperties.java b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaConsumerProperties.java index fec952db7..6c304ff4d 100644 --- a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaConsumerProperties.java +++ b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaConsumerProperties.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2017 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. @@ -21,6 +21,7 @@ import java.util.Map; /** * @author Marius Bogoevici + * @author Ilayaperumal Gopinathan * *

Thanks to Laszlo Szabo for providing the initial patch for generic property support.

*/ @@ -38,6 +39,8 @@ public class KafkaConsumerProperties { private boolean enableDlq; + private String dlqName; + private int recoveryInterval = 5000; private Map configuration = new HashMap<>(); @@ -119,4 +122,12 @@ public class KafkaConsumerProperties { public void setConfiguration(Map configuration) { this.configuration = configuration; } + + public String getDlqName() { + return dlqName; + } + + public void setDlqName(String dlqName) { + this.dlqName = dlqName; + } } diff --git a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java index c810f06b3..2d18792a4 100644 --- a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java +++ b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java @@ -55,6 +55,7 @@ import kafka.utils.ZkUtils; * * @author Soby Chacko * @author Gary Russell + * @author Ilayaperumal Gopinathan */ public class KafkaTopicProvisioner implements ProvisioningProvider, ExtendedProducerProperties>, InitializingBean { @@ -137,7 +138,8 @@ public class KafkaTopicProvisioner implements ProvisioningProvider.`. + By default, messages that result in errors will be forwarded to a topic named `error..`. + The DLQ topic name can be configurable via the property `dlqName`. This provides an alternative option to the more common Kafka replay scenario for the case when the number of errors is relatively small and replaying the entire original topic may be too cumbersome. + Default: `false`. @@ -166,6 +167,10 @@ configuration:: Map with a key/value pair containing generic Kafka consumer properties. + Default: Empty map. +dlqName:: + The name of the DLQ topic to receive the error messages. ++ +Default: null (If not specified, messages that result in errors will be forwarded to a topic named `error..`). === Kafka Producer Properties diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java index 84955306d..d2ec8e4ab 100644 --- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2016 the original author or authors. + * Copyright 2014-2017 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. @@ -201,7 +201,7 @@ public class KafkaMessageChannelBinder extends @Override @SuppressWarnings("unchecked") protected MessageProducer createConsumerEndpoint(final ConsumerDestination destination, final String group, - ExtendedConsumerProperties extendedConsumerProperties) { + final ExtendedConsumerProperties extendedConsumerProperties) { boolean anonymous = !StringUtils.hasText(group); Assert.isTrue(!anonymous || !extendedConsumerProperties.getExtension().isEnableDlq(), @@ -281,8 +281,9 @@ public class KafkaMessageChannelBinder extends : null; final byte[] payload = message.value() != null ? Utils.toArray(ByteBuffer.wrap((byte[]) message.value())) : null; - ListenableFuture> sentDlq = kafkaTemplate.send("error." + destination.getName() + "." + group, - message.partition(), key, payload); + String dlqName = StringUtils.hasText(extendedConsumerProperties.getExtension().getDlqName()) ? + extendedConsumerProperties.getExtension().getDlqName() : "error." + destination.getName() + "." + group; + ListenableFuture> sentDlq = kafkaTemplate.send(dlqName, message.partition(), key, payload); sentDlq.addCallback(new ListenableFutureCallback>() { StringBuilder sb = new StringBuilder().append(" a message with key='") .append(toDisplayString(ObjectUtils.nullSafeToString(key), 50)).append("'") diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java index de17ab083..03cf4ca2f 100644 --- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2017 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. @@ -274,6 +274,68 @@ public abstract class KafkaBinderTests extends PartitionCapableBinderTests producerProperties = createProducerProperties(); + producerProperties.setPartitionCount(10); + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + consumerProperties.setMaxAttempts(3); + consumerProperties.setBackOffInitialInterval(100); + consumerProperties.setBackOffMaxInterval(150); + consumerProperties.getExtension().setEnableDlq(true); + consumerProperties.getExtension().setAutoRebalanceEnabled(false); + String dlqName = "dlqTest"; + consumerProperties.getExtension().setDlqName(dlqName); + long uniqueBindingId = System.currentTimeMillis(); + Binding producerBinding = binder.bindProducer("retryTest." + uniqueBindingId + ".0", + moduleOutputChannel, producerProperties); + Binding consumerBinding = binder.bindConsumer("retryTest." + uniqueBindingId + ".0", + "testGroup", moduleInputChannel, consumerProperties); + ExtendedConsumerProperties dlqConsumerProperties = createConsumerProperties(); + dlqConsumerProperties.setMaxAttempts(1); + QueueChannel dlqChannel = new QueueChannel(); + Binding dlqConsumerBinding = binder.bindConsumer(dlqName, null, dlqChannel, dlqConsumerProperties); + + String testMessagePayload = "test." + UUID.randomUUID().toString(); + Message testMessage = MessageBuilder.withPayload(testMessagePayload).build(); + moduleOutputChannel.send(testMessage); + + Message dlqMessage = receive(dlqChannel, 3); + assertThat(dlqMessage).isNotNull(); + assertThat(dlqMessage.getPayload()).isEqualTo(testMessagePayload); + + // first attempt fails + assertThat(handler.getReceivedMessages().entrySet()).hasSize(1); + Message handledMessage = handler.getReceivedMessages().entrySet().iterator().next().getValue(); + assertThat(handledMessage).isNotNull(); + assertThat(handledMessage.getPayload()).isEqualTo(testMessagePayload); + assertThat(handler.getInvocationCount()).isEqualTo(consumerProperties.getMaxAttempts()); + binderBindUnbindLatency(); + dlqConsumerBinding.unbind(); + consumerBinding.unbind(); + + // on the second attempt the message is not redelivered because the DLQ is set + QueueChannel successfulInputChannel = new QueueChannel(); + consumerBinding = binder.bindConsumer("retryTest." + uniqueBindingId + ".0", "testGroup", + successfulInputChannel, consumerProperties); + String testMessage2Payload = "test." + UUID.randomUUID().toString(); + Message testMessage2 = MessageBuilder.withPayload(testMessage2Payload).build(); + moduleOutputChannel.send(testMessage2); + + Message receivedMessage = receive(successfulInputChannel); + assertThat(receivedMessage.getPayload()).isEqualTo(testMessage2Payload); + + binderBindUnbindLatency(); + consumerBinding.unbind(); + producerBinding.unbind(); + } + @Test @SuppressWarnings("unchecked") public void testAutoCreateTopicsEnabledSucceeds() throws Exception {