Remove sharing of the retrying field in FallbackBatchErrorHandler

[vooft]

This PR fixes 2 issues:

1. If an exception is thrown from ErrorHandlingUtils.retryBatch (for instance if SeekToCurrentBatchErrorHandler failed), then retrying field will never be reset back to false.
2. Error handlers are shared between different consumers when concurrency > 1, when one of them is retrying and rebalancing happens, it may result in other consumers being paused and never recovering again.

I'm not sure if it is a good idea in general to add a state to a shared class, it probably should be attributed to a particular consumer rather than an error handler.

Fixes #2382

[garyrussell]

Ouch! Good catch. Thanks.

Using a TL is fine since there is a 1:1 relationship between thread and consumer.

[vooft]

Fixed checkstyle

[vooft]

@garyrussell not sure if the failed test is related to my change, build is green on my machine.

KafkaMessageListenerContainerTests > testInvokeRecordInterceptorAllSkipped(AckMode, boolean) > org.springframework.kafka.listener.KafkaMessageListenerContainerTests.testInvokeRecordInterceptorAllSkipped(AckMode, boolean)[1] FAILED
    org.mockito.exceptions.verification.VerificationInOrderFailure: 
    Verification in order failure
    Wanted but not invoked:
    object.intercept(
        ConsumerRecord(topic = foo, partition = 0, leaderEpoch = null, offset = 1, NoTimestampType = -1, serialized key size = -1, serialized value size = -1, headers = RecordHeaders(headers = [], isReadOnly = false), key = 1, value = bar),
        Mock for Consumer, hashCode: 1199858180
    );
    -> at org.springframework.kafka.listener.KafkaMessageListenerContainerTests.testInvokeRecordInterceptorAllSkipped(KafkaMessageListenerContainerTests.java:3812)
    Wanted anywhere AFTER following interaction:
    consumer.commitSync(
        {foo-0 = OffsetAndMetadata{offset=1, leaderEpoch=null, metadata=''}},
        PT1M
    );
    -> at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doCommitSync(KafkaMessageListenerContainer.java:3189)
        at app//org.springframework.kafka.listener.KafkaMessageListenerContainerTests.testInvokeRecordInterceptorAllSkipped(KafkaMessageListenerContainerTests.java:3812)

[garyrussell]

Probably an unrelated flaky test. Will take a look on Monday.

[garyrussell]

Cherry-picked to 2.9.x, back ported to 2.8.x.

6a28cc5
3de1e89

Thanks @vooft - in future, please limit first commit headline to 50 chars.

[vooft]

@garyrussell thank you! Sure, will keep it short next time.