Prevent message being reprocessed endlessly many times if using isStopContainerWhenFenced and message cannot be processed because of transaction timeout

[bartosz-stasikowski-projectdrgn]

Affects Version(s): 2.6.9
🎁 Enhancement (proposal)

Context: non batch transaction processing

As we know ProducerFencedException can be caused by producer being fenced (in this case we assume that producers listener is no longer assigned to the same partition) or it is transaction timeout (and https://issues.apache.org/jira/browse/KAFKA-9803 is still not done).
I saw that to handle it somehow after discussion in #1612, container can be stopped (by setting isStopContainerWhenFenced to true)

This is ok but it can cause that some poison pill message (because of poorly written listener) that is timing out transaction every time will stop containers on all pods.
Even if there is some spring listener which is starting those containers back or pods are restarted this message wont be processed.

Any change to handle it somehow? For example track records which are fenced and do not process them again after container is restarted (if they could not be processed configured number of times).
This is the only solution that came to my mind, since we cannot do any rollback processing when producer was fenced.

[garyrussell]

This would be tricky to implement and would require major changes; the container currently maintains no previous state when it stopped (a completely new container is created each time).

Need to give it some thought.