GH-2340: Fix Retrying Batch Error Handling #2341
Conversation
Resolves spring-projects#2340 The `RetryingBatchErrorHandler` - now called the `FallbackBatchErrorHandler` pauses and resumes the consumer during retries, to allow it to poll the consumer to avoid a forced rebalance. However, if a normal rebalance occurs, for example if a new member joins, the error handler does not re-pause the consumer and silently consumes new records. Add a mechanism to always re-pause the consume when in this retry mode. **cherry-pick to 2.9.x, 2.8.x**
|
|
||
| private boolean ackAfterHandle = true; | ||
|
|
||
| private boolean retrying; |
There was a problem hiding this comment.
No need; always used from the consumer thread.
There was a problem hiding this comment.
So, you mean that onPartitionsAssigned() is called withing that ErrorHandlingUtils.retryBatch() process?
Otherwise I don't see how this variable is involved...
Thanks
There was a problem hiding this comment.
ErrorHandlingUtils.retryBatch pauses the consumer and polls in a while loop until retries are exhausted, and then resumes.
We need to repause the consumer if onPartitionsAssigned() is called while we are in that loop (it is called from poll() in the kafka-clients).
So the thread that tests this field is the same thread that is looping within retryBatch.
|
... and cherry-picked to It does not apply clearly to Thanks |
|
Back-ported to 2.8.x |
2 participants
Resolves #2340
The
RetryingBatchErrorHandler- now called theFallbackBatchErrorHandlerpauses and resumes the consumer during retries, to allow it to poll the
consumer to avoid a forced rebalance.
However, if a normal rebalance occurs, for example if a new member joins,
the error handler does not re-pause the consumer and silently consumes
new records.
Add a mechanism to always re-pause the consume when in this retry mode.
cherry-pick to 2.9.x, 2.8.x
As well as the unit test, tested proper functionality by running two copies of a Boot app.