diff --git a/docs/src/main/asciidoc/dlq.adoc b/docs/src/main/asciidoc/dlq.adoc index 2e922ecc5..a8bf728d6 100644 --- a/docs/src/main/asciidoc/dlq.adoc +++ b/docs/src/main/asciidoc/dlq.adoc @@ -22,6 +22,8 @@ public DlqPartitionFunction partitionFunction() { ---- ==== +NOTE: If you set a consumer binding's `dlqPartitions` property to 1 (and the binder's `minPartitionCount` is equal to `1`), there is no need to supply a `DlqPartitionFunction`; the framework will always use partition 0. +If you set a consumer binding's `dlqPartitions` property to a value greater than `1` (or the binder's `minPartitionCount` is greater than `1`), you **must** provide a `DlqPartitionFunction` bean, even if the partition count is the same as the original topic's. [[dlq-handling]] ==== Handling Records in a Dead-Letter Topic diff --git a/docs/src/main/asciidoc/overview.adoc b/docs/src/main/asciidoc/overview.adoc index 828e719a2..d2749b855 100644 --- a/docs/src/main/asciidoc/overview.adoc +++ b/docs/src/main/asciidoc/overview.adoc @@ -213,6 +213,15 @@ See <> for how to change that behavior. **Not allowed when `destinationIsPattern` is `true`.** + Default: `false`. +dlqPartitions:: +When `enableDlq` is true, and this property is not set, a dead letter topic with the same number of partitions as the primary topic(s) is created. +Usually, dead-letter records are sent to the same partition in the dead-letter topic as the original record. +This behavior can be changed; see <>. +If this property is set to `1` and there is no `DqlPartitionFunction` bean, all dead-letter records will be written to partition `0`. +If this property is greater than `1`, you **MUST** provide a `DlqPartitionFunction` bean. +Note that the actual partition count is affected by the binder's `minPartitionCount` property. ++ +Default: `none` configuration:: Map with a key/value pair containing generic Kafka consumer properties. In addition to having Kafka consumer properties, other configuration properties can be passed here. 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 1e08f4012..1039c278c 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 @@ -100,6 +100,8 @@ public class KafkaConsumerProperties { private String dlqName; + private Integer dlqPartitions; + private KafkaProducerProperties dlqProducerProperties = new KafkaProducerProperties(); private int recoveryInterval = 5000; @@ -215,6 +217,14 @@ public class KafkaConsumerProperties { this.dlqName = dlqName; } + public Integer getDlqPartitions() { + return this.dlqPartitions; + } + + public void setDlqPartitions(Integer dlqPartitions) { + this.dlqPartitions = dlqPartitions; + } + public String[] getTrustedPackages() { return this.trustedPackages; } 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 500fb88b0..8de9cb40a 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 @@ -274,12 +274,16 @@ public class KafkaTopicProvisioner implements private ConsumerDestination createDlqIfNeedBe(AdminClient adminClient, String name, String group, ExtendedConsumerProperties properties, boolean anonymous, int partitions) { + if (properties.getExtension().isEnableDlq() && !anonymous) { String dlqTopic = StringUtils.hasText(properties.getExtension().getDlqName()) ? properties.getExtension().getDlqName() : "error." + name + "." + group; + int dlqPartitions = properties.getExtension().getDlqPartitions() == null + ? partitions + : properties.getExtension().getDlqPartitions(); try { - createTopicAndPartitions(adminClient, dlqTopic, partitions, + createTopicAndPartitions(adminClient, dlqTopic, dlqPartitions, properties.getExtension().isAutoRebalanceEnabled(), properties.getExtension().getTopic()); } diff --git a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/utils/DlqPartitionFunction.java b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/utils/DlqPartitionFunction.java index f3025302e..8ed0534e2 100644 --- a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/utils/DlqPartitionFunction.java +++ b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/utils/DlqPartitionFunction.java @@ -16,6 +16,7 @@ package org.springframework.cloud.stream.binder.kafka.utils; +import org.apache.commons.logging.Log; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.springframework.lang.Nullable; @@ -32,6 +33,16 @@ import org.springframework.lang.Nullable; @FunctionalInterface public interface DlqPartitionFunction { + /** + * Returns the same partition as the original recor. + */ + DlqPartitionFunction ORIGINAL_PARTITION = (group, rec, ex) -> rec.partition(); + + /** + * Returns 0. + */ + DlqPartitionFunction PARTITION_ZERO = (group, rec, ex) -> 0; + /** * Apply the function. * @param group the consumer group. @@ -42,4 +53,24 @@ public interface DlqPartitionFunction { @Nullable Integer apply(String group, ConsumerRecord record, Throwable throwable); + /** + * Determine the fallback function to use based on the dlq partition count if no + * {@link DlqPartitionFunction} bean is provided. + * @param dlqPartitions the partition count. + * @param logger the logger. + * @return the fallback. + */ + static DlqPartitionFunction determineFallbackFunction(@Nullable Integer dlqPartitions, Log logger) { + if (dlqPartitions == null) { + return ORIGINAL_PARTITION; + } + else if (dlqPartitions > 1) { + logger.error("'dlqPartitions' is > 1 but a custom DlqPartitionFunction bean is not provided"); + return ORIGINAL_PARTITION; + } + else { + return PARTITION_ZERO; + } + } + } diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderUtils.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderUtils.java index 8816d0c88..33b2b9e1b 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderUtils.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderUtils.java @@ -20,6 +20,8 @@ import java.util.HashMap; import java.util.Map; import java.util.function.BiFunction; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.common.TopicPartition; @@ -52,6 +54,8 @@ import org.springframework.util.StringUtils; */ final class KafkaStreamsBinderUtils { + private static final Log LOGGER = LogFactory.getLog(KafkaStreamsBinderUtils.class); + private KafkaStreamsBinderUtils() { } @@ -78,9 +82,11 @@ final class KafkaStreamsBinderUtils { Map partitionFunctions = context.getBeansOfType(DlqPartitionFunction.class, false, false); - DlqPartitionFunction partitionFunction = partitionFunctions.size() == 1 + boolean oneFunctionPresent = partitionFunctions.size() == 1; + Integer dlqPartitions = extendedConsumerProperties.getExtension().getDlqPartitions(); + DlqPartitionFunction partitionFunction = oneFunctionPresent ? partitionFunctions.values().iterator().next() - : (grp, rec, ex) -> rec.partition(); + : DlqPartitionFunction.determineFallbackFunction(dlqPartitions, LOGGER); ProducerFactory producerFactory = getProducerFactory( new ExtendedProducerProperties<>( diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/DeserializtionErrorHandlerByBinderTests.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/DeserializtionErrorHandlerByBinderTests.java index fef08cbfd..cb81eeb55 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/DeserializtionErrorHandlerByBinderTests.java +++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/DeserializtionErrorHandlerByBinderTests.java @@ -64,6 +64,7 @@ public abstract class DeserializtionErrorHandlerByBinderTests { @ClassRule public static EmbeddedKafkaRule embeddedKafkaRule = new EmbeddedKafkaRule(1, true, + "foos", "counts-id", "error.foos.foobar-group", "error.foos1.fooz-group", "error.foos2.fooz-group"); @@ -112,19 +113,19 @@ public abstract class DeserializtionErrorHandlerByBinderTests { "spring.cloud.stream.kafka.streams.binder.serdeError=sendToDlq", "spring.cloud.stream.kafka.streams.bindings.input.consumer.application-id" + "=deserializationByBinderAndDlqTests", + "spring.cloud.stream.kafka.streams.bindings.input.consumer.dlqPartitions=1", "spring.cloud.stream.bindings.input.group=foobar-group" }, webEnvironment = SpringBootTest.WebEnvironment.NONE) public static class DeserializationByBinderAndDlqTests extends DeserializtionErrorHandlerByBinderTests { @Test - @SuppressWarnings("unchecked") public void test() { Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>( senderProps); KafkaTemplate template = new KafkaTemplate<>(pf, true); template.setDefaultTopic("foos"); - template.sendDefault(7, "hello"); + template.sendDefault(1, 7, "hello"); Map consumerProps = KafkaTestUtils.consumerProps("foobar", "false", embeddedKafka); @@ -137,7 +138,8 @@ public abstract class DeserializtionErrorHandlerByBinderTests { ConsumerRecord cr = KafkaTestUtils.getSingleRecord(consumer1, "error.foos.foobar-group"); - assertThat(cr.value().equals("hello")).isTrue(); + assertThat(cr.value()).isEqualTo("hello"); + assertThat(cr.partition()).isEqualTo(0); // Ensuring that the deserialization was indeed done by the binder verify(conversionDelegate).deserializeOnInbound(any(Class.class), 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 5627f2e15..01277a54a 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 @@ -246,7 +246,7 @@ public class KafkaMessageChannelBinder extends this.rebalanceListener = rebalanceListener; this.dlqPartitionFunction = dlqPartitionFunction != null ? dlqPartitionFunction - : (group, rec, ex) -> rec.partition(); + : null; } private static String[] headersToMap( @@ -967,6 +967,7 @@ public class KafkaMessageChannelBinder extends protected MessageHandler getErrorMessageHandler(final ConsumerDestination destination, final String group, final ExtendedConsumerProperties properties) { + KafkaConsumerProperties kafkaConsumerProperties = properties.getExtension(); if (kafkaConsumerProperties.isEnableDlq()) { KafkaProducerProperties dlqProducerProperties = kafkaConsumerProperties @@ -1083,12 +1084,21 @@ public class KafkaMessageChannelBinder extends ? kafkaConsumerProperties.getDlqName() : "error." + record.topic() + "." + group; dlqSender.sendToDlq(recordToSend.get(), kafkaHeaders, dlqName, group, throwable, - this.dlqPartitionFunction); + determinDlqPartitionFunction(properties.getExtension().getDlqPartitions())); }; } return null; } + private DlqPartitionFunction determinDlqPartitionFunction(Integer dlqPartitions) { + if (this.dlqPartitionFunction != null) { + return this.dlqPartitionFunction; + } + else { + return DlqPartitionFunction.determineFallbackFunction(dlqPartitions, this.logger); + } + } + @Override protected MessageHandler getPolledConsumerErrorMessageHandler( ConsumerDestination destination, String group, 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 07c241062..a0aa330d0 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 @@ -859,33 +859,44 @@ public class KafkaBinderTests extends @Test public void testDlqAndRetry() throws Exception { - testDlqGuts(true, null); + testDlqGuts(true, null, null); } @Test public void testDlq() throws Exception { - testDlqGuts(false, null); + testDlqGuts(false, null, 3); } @Test public void testDlqNone() throws Exception { - testDlqGuts(false, HeaderMode.none); + testDlqGuts(false, HeaderMode.none, 1); } @Test public void testDlqEmbedded() throws Exception { - testDlqGuts(false, HeaderMode.embeddedHeaders); + testDlqGuts(false, HeaderMode.embeddedHeaders, 3); } - private void testDlqGuts(boolean withRetry, HeaderMode headerMode) throws Exception { + private void testDlqGuts(boolean withRetry, HeaderMode headerMode, Integer dlqPartitions) throws Exception { + int expectedDlqPartition = dlqPartitions == null ? 0 : dlqPartitions - 1; KafkaBinderConfigurationProperties binderConfig = createConfigurationProperties(); - binderConfig.setMinPartitionCount(2); - AbstractKafkaTestBinder binder = getBinder(binderConfig, (group, rec, ex) -> 0); + DlqPartitionFunction dlqPartitionFunction; + if (Integer.valueOf(1).equals(dlqPartitions)) { + dlqPartitionFunction = null; // test that ZERO_PARTITION is used + } + else if (dlqPartitions == null) { + dlqPartitionFunction = (group, rec, ex) -> 0; + } + else { + dlqPartitionFunction = (group, rec, ex) -> dlqPartitions - 1; + } + AbstractKafkaTestBinder binder = getBinder(binderConfig, dlqPartitionFunction); ExtendedProducerProperties producerProperties = createProducerProperties(); producerProperties.getExtension() .setHeaderPatterns(new String[] { MessageHeaders.CONTENT_TYPE }); producerProperties.setHeaderMode(headerMode); + producerProperties.setPartitionCount(2); DirectChannel moduleOutputChannel = createBindableChannel("output", createProducerBindingProperties(producerProperties)); @@ -898,6 +909,8 @@ public class KafkaBinderTests extends consumerProperties.getExtension().setAutoRebalanceEnabled(false); consumerProperties.setHeaderMode(headerMode); consumerProperties.setMultiplex(true); + consumerProperties.getExtension().setDlqPartitions(dlqPartitions); + consumerProperties.setConcurrency(2); DirectChannel moduleInputChannel = createBindableChannel("input", createConsumerBindingProperties(consumerProperties)); @@ -917,9 +930,20 @@ public class KafkaBinderTests extends MessageListenerContainer container = TestUtils.getPropertyValue(consumerBinding, "lifecycle.messageListenerContainer", MessageListenerContainer.class); - assertThat(container.getContainerProperties().getTopicPartitions().length) + assertThat(container.getContainerProperties().getTopicPartitionsToAssign().length) .isEqualTo(4); // 2 topics 2 partitions each + try (AdminClient admin = AdminClient.create(Collections.singletonMap(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, + embeddedKafka.getEmbeddedKafka().getBrokersAsString()))) { + + Map topicDescriptions = admin.describeTopics(Collections.singletonList("error.dlqTest." + uniqueBindingId + ".0.testGroup")) + .all() + .get(10, TimeUnit.SECONDS); + assertThat(topicDescriptions).hasSize(1); + assertThat(topicDescriptions.values().iterator().next().partitions()) + .hasSize(dlqPartitions == null ? 2 : dlqPartitions); + } + ExtendedConsumerProperties dlqConsumerProperties = createConsumerProperties(); dlqConsumerProperties.setMaxAttempts(1); dlqConsumerProperties.setHeaderMode(headerMode); @@ -985,7 +1009,7 @@ public class KafkaBinderTests extends assertThat(receivedMessage.getHeaders() .get(KafkaMessageChannelBinder.X_EXCEPTION_FQCN)).isNotNull(); assertThat(receivedMessage.getHeaders() - .get(KafkaHeaders.RECEIVED_PARTITION_ID)).isEqualTo(0); + .get(KafkaHeaders.RECEIVED_PARTITION_ID)).isEqualTo(expectedDlqPartition); } else if (!HeaderMode.none.equals(headerMode)) { assertThat(handler.getInvocationCount()) @@ -1022,7 +1046,7 @@ public class KafkaBinderTests extends .get(KafkaMessageChannelBinder.X_EXCEPTION_FQCN)).isNotNull(); assertThat(receivedMessage.getHeaders() - .get(KafkaHeaders.RECEIVED_PARTITION_ID)).isEqualTo(0); + .get(KafkaHeaders.RECEIVED_PARTITION_ID)).isEqualTo(expectedDlqPartition); } else { assertThat(receivedMessage.getHeaders()