KAFKA-9734: Fix IllegalState in Streams transit to standby #8319
Conversation
There was a problem hiding this comment.
Hey @guozhangwang ,
What do you think of this fix? I tried for quite a while to add a regression test for this, but it's quite tricky because the bug requires a specific interleaving of the Producer, StreamThread, and TaskManager. I ultimately decided such a test would be pointless anyway, since all it would be exercising a code path which this PR removes.
I also didn't add a new test because the existing tests already cover the transitions in question. It's just that a specific edge case no longer exists.
Does that make sense, or am I cheating?
Thanks,
-John
| // pause all partitions that are for standby tasks from the restore consumer | ||
| pauseChangelogsFromRestoreConsumer(standbyRestoringChangelogs()); | ||
| // pause all partitions that are for standby tasks from the restore consumer | ||
| pauseChangelogsFromRestoreConsumer(standbyRestoringChangelogs()); |
There was a problem hiding this comment.
I moved this inside the block because we should only need to do it if we are transitioning from restoring standbys. I wanted to make the idempotent call slightly more efficient, since we now call this method every time in StreamThread#runOnce.
There was a problem hiding this comment.
We can move the line 290 inside as well?
There was a problem hiding this comment.
Oh, yeah, of course :)
| throw new IllegalStateException("The changelog reader is not restoring active tasks while trying to " + | ||
| "transit to update standby tasks: " + changelogs); | ||
| throw new IllegalStateException( | ||
| "The changelog reader is not restoring active tasks (is " + state + ") while trying to " + |
There was a problem hiding this comment.
This might be a bit redundant, but it wasted a bit of my time to find out that if it wasn't in active_restoring, it must have been in restoring standbys because that's the only other state. I figure it doesn't hurt to just be explicit.
| log.warn("Detected the states of tasks " + e.corruptedTaskWithChangelogs() + " are corrupted. " + | ||
| "Will close the task as dirty and re-create and bootstrap from scratch.", e); |
There was a problem hiding this comment.
Rearranged this so we can include the exception itself. The stacktrace is useful for tracking down the reason for the exception.
| "This implies that this thread missed a rebalance and dropped out of the consumer group. " + | ||
| "Will close out all assigned tasks and rejoin the consumer group."); | ||
| "This implies that this thread missed a rebalance and dropped out of the consumer group. " + | ||
| "Will close out all assigned tasks and rejoin the consumer group.", e); |
| @@ -578,13 +578,13 @@ void runOnce() { | |||
| // only try to initialize the assigned tasks | |||
| // if the state is still in PARTITION_ASSIGNED after the poll call | |||
| if (state == State.PARTITIONS_ASSIGNED) { | |||
| // transit to restore active is idempotent so we can call it multiple times | |||
| changelogReader.enforceRestoreActive(); | |||
There was a problem hiding this comment.
This is the primary fix. Instead of relying (hoping) on TaskManager to put the changelog reader into restoring_active, we just idempotently make sure it's in that state any time we're in partitions_assigned.
| @@ -279,13 +279,13 @@ private boolean hasRestoredToEnd(final ChangelogMetadata metadata) { | |||
| // NOTE: even if the newly created tasks do not need any restoring, we still first transit to this state and then | |||
| // immediately transit back -- there's no overhead of transiting back and forth but simplifies the logic a lot. | |||
| @Override | |||
| public void transitToRestoreActive() { | |||
| public void enforceRestoreActive() { | |||
There was a problem hiding this comment.
Renamed this method to make it clear we aren't necessarily "transitioning", we actually call it all the time now any time we want to "be in restoreActive".
| changeLogReader.transitToRestoreActive(); | ||
| expectLastCall(); |
There was a problem hiding this comment.
This is a strict mock, so we have to remove this expectation, since the TaskManager no longer calls it.
There was a problem hiding this comment.
Just one nit, otherwise LGTM! Thanks @vvcephei
Please feel free to merge after green builds.
| // pause all partitions that are for standby tasks from the restore consumer | ||
| pauseChangelogsFromRestoreConsumer(standbyRestoringChangelogs()); | ||
| // pause all partitions that are for standby tasks from the restore consumer | ||
| pauseChangelogsFromRestoreConsumer(standbyRestoringChangelogs()); |
There was a problem hiding this comment.
We can move the line 290 inside as well?
|
This time, there were two different, unrelated failures. kafka.api.PlaintextConsumerTest.testLowMaxFetchSizeForRequestAndPartition |
Consolidate ChangelogReader state management inside of StreamThread to avoid having to reason about all execution paths in both StreamThread and TaskManager.
Committer Checklist (excluded from commit message)