GH-628: Add dlqPartitions property

Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/628

Allow users to specify the number of partitions in the dead letter topic.

If the property is set to 1 and the `minPartitionCount` binder property is 1,
override the default behavior and always publish to partition 0.

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

docs/src/main/asciidoc/overview.adoc

@@ -213,6 +213,15 @@ See <<dlq-partition-selection>> 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 <<dlq-partition-selection>>.
+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.

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;
 	}

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

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;
+		}
+	}
+
 }

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<String, DlqPartitionFunction> 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<byte[], byte[]> producerFactory = getProducerFactory(
 					new ExtendedProducerProperties<>(

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<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
 			DefaultKafkaProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(
 					senderProps);
 			KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf, true);
 			template.setDefaultTopic("foos");
-			template.sendDefault(7, "hello");
+			template.sendDefault(1, 7, "hello");
 
 			Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("foobar",
 					"false", embeddedKafka);
@@ -137,7 +138,8 @@ public abstract class DeserializtionErrorHandlerByBinderTests {
 
 			ConsumerRecord<String, String> 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),

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<KafkaConsumerProperties> 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,

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<KafkaProducerProperties> 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<String, TopicDescription> 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<KafkaConsumerProperties> 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()