Default destination resolver in DeadLetterPublishingRecoverer issue when DLT partitions are less than original topic partitions

[ecapoccia]

The default destination resolver in the DeadLetterPublishingRecoverer routes the records to a topic partition with the same name as the incoming consumer record + the suffix ".DLT", and the same partition as the incoming consumer record.

DEFAULT_DESTINATION_RESOLVER = (cr, e) -> new TopicPartition(cr.topic() + ".DLT", cr.partition());

This is a problem as this logic assumes that the number of partitions for the DLT is the same as the number of partitions of the original topic, which is not necessarily true.
In case the partitions for the DLT are less, this result in exceptions, as the destination partitions do not exist. Likewise in the (unlikely) case the DTL has more partitions than the original topic, some of them will be idle.

The solution I have successfully used is to set the destination partition to -1, which will delegate the choice of the destination partition within the DLT to Kafka.

DEFAULT_DESTINATION_RESOLVER = (cr, e) -> new TopicPartition(cr.topic() + ".DLT", -1);

Will submit a PR with a test to prove the case. I think this should be the default mechanism.

[garyrussell]

As I said on the PR; we can't just change the default like that; we will make the documentation more visible,

[garyrussell]

On reflection, I think we could inspect the DLT (if we have permissions) and log a WARN if the resolved partition doesn't exist.

That will satisfy your request, and prevent a breaking behavior change.