Record recovery in transactional batch listener

[tommyk-gears]

I am trying to implement/configure error handling for Kafka listeners with two relatively straight forward basic requirements;

1. Records that can be successfully processed should be successfully processed.
2. Records that can not be successfully processed should be sent to a record recoverer.

I'd assume those are quite common/universal requirements.

For now I'm assuming that any batch listener will throw BatchListenerFailedException to indicate which record failed.

I have four different cases that I want to support;

1. Non-transactional record listener
2. Non-transactional batch listener
3. Transactional record listener
4. Transactional batch listener

Now, I managed to get the two first cases working by using a CommonErrorHandler which sends failing records to a record recoverer, and the third case also works by using DefaultAfterRollbackProcessor which sends the failing record to a record recoverer.

However, the fourth case seems a whole lot harder. As I understand it the batch is "atomic" when it comes to recovery, either you recover all records in a failing batch, or you redeliver the complete batch. I tried using a method interceptor along the lines of this comment: #2233 (comment), which lets me retry partial batches - but I fail to figure out how to commit offsets for successfully retried partial batches. The offset commit logics is very much in the hands of the ListenerContainer, which leads me to believe that the retry logics also needs to be inside the ListenerContainer. The logics I'd like to have here is basically;

• Catch BatchListenerFailedException and get the index of the failing record.
• Redeliver a new batch with all records preceding the failing records (we can assume that this batch will be successfully processed and can be committed)
• Redeliver a batch with only the failing record (we can assume that this batch will fail)
• Redeliver a batch with any remaining records from the original batch

Now we can have a AfterRollbackProcessor that will recover and commit recovered batches only if the batch contains a single record.

Does this make sense? Or is there a way to have this functionality with existing abstractions (I considered using a BatchMessageListener adapter, but I don't see a way forward with it)?

Also, it is mentioned in the documentation that "Default behavior, recovery is not possible with a batch listener, since the framework has no knowledge about which record in the batch keeps failing". Is this really true? There is nothing stopping a transactional batch listener from using BatchListenerFailedException to find out which record failed as far as I understand.

[sobychacko]

@tommyk-gears Can you refine your exact use case? The way I understand the issue is that you have a transactional batch listener and want to recover failing records. Is that the right assumption? By default, you can only recover the entire batch. Upon failure of a record, the entire batch will be redelivered. We made some recent improvements in this area. Can you take a look at this issue and the associated PR to see if the improvements mentioned here give you any pointers? I am not suggesting that these are the solutions for your issue, but I want to start with these recent changes and see if we can get to a point where we can isolate the exact use case at hand. Thanks!

[tommyk-gears]

Yes exactly, in a transactional batch listener i want to recover only the failing record(s).
I tried out the batchRecoverAfterRollback feature from #2588 but it allows me to recover the entire batch atomically only.

It is mentioned in this comment that DefaultAfterRollbackProcessor probably should understand BatchListenerFailedException and only recover the failing record. If this is possible it would be a really nice addition to DefaultAfterRollbackProcessor imo.

[sobychacko]

We will look into it further and get back to you. By the way, it's holidays and summer vacations, etc., so expect a delayed update, but we will make all the efforts to triage it. Thanks!