Custom DLQ Destination Resolver (#976)

* Custom DLQ Destination Resolver

Allow applications to provide a custom DLQ destination resolver
implementaiton by providing a new interface DlqDestinationResolver
as part of binder's public contract. This interface is a BiFunction
extension using which the applications can provide more fine grained
control over where to route records in error.

Adding test to verify.

Adding docs.

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

* Add DlqDestinationResolver to MessageChannel based binder.

Tests and docs

docs/src/main/asciidoc/dlq.adoc

@@ -21,10 +21,36 @@ 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.
 
+It is also possible to define a custom name for the DLQ topic.
+In order to do so, create an implementation of `DlqDestinationResolver` as a `@Bean` to the application context.
+When the binder detects such a bean, that takes precedence, otherwise it will use the `dlqName` property.
+If neither of these are found, it will default to `error.<destination>.<group>`.
+Here is an example of `DlqDestinationResolver` as a `@Bean`.
+
+====
+[source]
+----
+@Bean
+public DlqDestinationResolver dlqDestinationResolver() {
+    return (rec, ex) -> {
+        if (rec.topic().equals("word1")) {
+            return "topic1-dlq";
+        }
+        else {
+            return "topic2-dlq";
+        }
+    };
+}
+----
+====
+
+One important thing to keep in mind when providing an implementation for `DlqDestinationResolver` is that the provisioner in the binder will not auto create topics for the application.
+This is because there is no way for the binder to infer the names of all the DLQ topics the implementation might send to.
+Therefore, if you provide DLQ names using this strategy, it is the application's responsibility to ensure that those topics are created beforehand.
+
 [[dlq-handling]]
 ==== Handling Records in a Dead-Letter Topic
 

docs/src/main/asciidoc/kafka-streams.adoc

@@ -832,13 +832,41 @@ When the above property is set, all the records in deserialization error are aut
 
 You can set the topic name where the DLQ messages are published as below.
 
+You can provide an implementation for `DlqDestinationResolver` which is a functional interface.
+`DlqDestinationResolver` takes `ConsumerRecord` and the exception as inputs and then allows to specify a topic name as the output.
+By gaining access to the Kafka `ConsumerRecord`, the header records can be introspected in the implementation of the `BiFunction`.
+
+Here is an example of providing an implementation for `DlqDestinationResolver`.
+
+[source]
+----
+@Bean
+public DlqDestinationResolver dlqDestinationResolver() {
+    return (rec, ex) -> {
+        if (rec.topic().equals("word1")) {
+            return "topic1-dlq";
+        }
+        else {
+            return "topic2-dlq";
+        }
+    };
+}
+----
+
+One important thing to keep in mind when providing an implementation for `DlqDestinationResolver` is that the provisioner in the binder will not auto create topics for the application.
+This is because there is no way for the binder to infer the names of all the DLQ topics the implementation might send to.
+Therefore, if you provide DLQ names using this strategy, it is the application's responsibility to ensure that those topics are created beforehand.
+
+If `DlqDestinationResolver` is present in the application as a bean, that takes higher prcedence.
+If you do not want to follow this approach and rather provide a static DLQ name using configuration, you can set the following property.
+
 [source]
 ----
 spring.cloud.stream.kafka.streams.bindings.process-in-0.consumer.dlqName: custom-dlq (Change the binding name accordingly)
 ----
 
 If this is set, then the error records are sent to the topic `custom-dlq`.
-If this is not set, then it will create a DLQ topic with the name `error.<input-topic-name>.<application-id>`.
+If the application is not using either of the above strategies, then it will create a DLQ topic with the name `error.<input-topic-name>.<application-id>`.
 For instance, if your binding's destination topic is `inputTopic` and the application ID is `process-applicationId`, then the default DLQ topic is `error.inputTopic.process-applicationId`.
 It is always recommended to explicitly create a DLQ topic for each input binding if it is your intention to enable DLQ.
 

docs/src/main/asciidoc/overview.adoc

@@ -220,7 +220,7 @@ Default: null (equivalent to `earliest`).
 enableDlq::
 When set to true, it enables DLQ behavior for the consumer.
 By default, messages that result in errors are forwarded to a topic named `error.<destination>.<group>`.
-The DLQ topic name can be configurable by setting the `dlqName` property.
+The DLQ topic name can be configurable by setting the `dlqName` property or by defining a `@Bean` of type `DlqDestinationResolver`.
 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.
 See <<kafka-dlq-processing>> processing for more information.
 Starting with version 2.0, messages sent to the DLQ topic are enhanced with the following headers: `x-original-topic`, `x-exception-message`, and `x-exception-stacktrace` as `byte[]`.