KAFKA-19831: Improved error handling in DefaultStateUpdater. (#20767) #20958
Conversation
…20767) - Improved error handling in DefaultStateUpdater to take potential failures in Task#maybeCheckpoint into account. - Improved TaskManager#shutdownStateUpdater to not hang indefinitely if the State Updater thread is dead. Reviewers: Matthias J. Sax <[email protected]>, Lucas Brutschy <[email protected]> --------- Co-authored-by: Matthias J. Sax <[email protected]>
|
|
||
| streams.removeStreamThread(); | ||
|
|
||
| TestUtils.waitForCondition(() -> streams.state() == KafkaStreams.State.REBALANCING, TimeUnit.MINUTES.toMillis(2), "Streams never reached REBALANCING state"); |
There was a problem hiding this comment.
Why do we need this only for 4.1 branch?
There was a problem hiding this comment.
for some reason in 4.1 branch rebalancing kicks in only after ~1 minute after the thread is removed. it didn't happen in trunk
There was a problem hiding this comment.
Ah. Interesting. Did dig into this and found the root cause: ebae768#diff-ab27af136b0c45ed402ec44368a91380b018bf06f1a9722324fe6be8d5220f7dR1204
We can keep this condition (maybe reduce to 1 minute waiting time?), and additionally reduce session timeout to 10 sec to speed up the test?
W/o sending a leave group request, the test hangs waiting for session timeout to expire on missing heartbeats from the removed thread.
There was a problem hiding this comment.
Thanks a lot! it explains everything. updated the PR
2 participants
failures in Task#maybeCheckpoint into account.
the State Updater thread is dead.
Cherry-pick of #20767
Reviewers: Matthias J. Sax [email protected]