Multi consumer (group) support SeekToCurrentErrorHandler, FailedRecordTracker & DeadLetterPublishingRecoverer?

[hartmut-co-uk]

Affects Version(s): Tested on 2.2.3.RELEASE (but probably 2.2.*)

🎁 Enhancement
Hi,

While exploring and trialing the new spring-kafka 2.2 SeekToCurrentErrorHandler & DeadLetterPublishingRecoverer 💪 I came across a limitation with the recoverer in regards to working with multiple consumers (different consumer-groups) processing the same topic.
Neither can the consumer-group be retrieved nor is the information added via the enrichment to the message headers (KafkaHeaders.DLT_*).

Basically for consumers CA & CB consuming from topic T1 - I'd like to use separate dead-letter queues, one by consumer-group.
E.g. if CA succeeds to process the message - while CB fails due to say a downstream system outage -> timeout -> I'd like to publish the message to a dead-letter topic T1.CB.DLT - to be able to identify which consumer failed processing the message and handle accordingly.

This doesn't seem to be doable with the given decorator interface BiConsumer<ConsumerRecord<?, ?>, Exception> / BiFunction<ConsumerRecord<?, ?>, Exception, TopicPartition> destinationResolver, correct?

Would there be a way/option to pass information on the listener / consumer-group (groupId) down to the DeadLetterPublishingRecoverer to allow such a setup?

[garyrussell]

Sounds like a reasonable request; we'll look into it.

I guess the only work around is to configure a different error handler instance into each container factory.

[hartmut-co-uk]

👍 and thanks for the the workaround suggestion. I've got that setup working - but I really end up with very verbose configuration of container factories 😁
It's now essentially a product of event-type x consumer-group - as I've typed consumer factories with defined typed JsonDeserializer and the consumer-group name in 2 places now (config -> container factory & @KafkaListener(groupId = ...))

[garyrussell]

I have taken a quick look and I think the simplest solution is to add a groupId property to the ListenerExecutionFailedException - that way it will propagate all the way to the recoverer without any API changes.

[hartmut-co-uk]

Sounds good! Thanks a lot for the quick response!