[
https://issues.apache.org/jira/browse/KAFKA-20731?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18091985#comment-18091985
]
Lucas Brutschy commented on KAFKA-20731:
----------------------------------------
We have to ask ourselves about the invariant: what is the intended state when
there are zero updating tasks? The problem with the code seems to be, with zero
updating tasks we can either be in STANDBY_UPDATING or ACTIVE_RESTORING - there
is no invariant. However, the right invariant is clearly stated in
StoreChangelogReader: // A changelog reader with no tasks at all is in state
ACTIVE_RESTORING. Which would mean we have to transition to ACTIVE_RESTORING
when the last task is paused or removed?
> Streams: transitToUpdateStandby is called without checking ACTIVE_RESTORING,
> crashing the state updater thread
> --------------------------------------------------------------------------------------------------------------
>
> Key: KAFKA-20731
> URL: https://issues.apache.org/jira/browse/KAFKA-20731
> Project: Kafka
> Issue Type: Bug
> Components: streams
> Affects Versions: 4.3.0, 4.1.2
> Reporter: Eswarar Siva
> Priority: Minor
>
> The state updater thread can die with (captured on 4.3.0):
> java.lang.IllegalStateException: The changelog reader is not restoring active
> tasks (is STANDBY_UPDATING) while trying to
> transit to update standby tasks: {...}
> at
> org.apache.kafka.streams.processor.internals.StoreChangelogReader.transitToUpdateStandby(StoreChangelogReader.java:326)
> at
> org.apache.kafka.streams.processor.internals.DefaultStateUpdater$StateUpdaterThread.resumeTask(DefaultStateUpdater.java:661)
> at
> org.apache.kafka.streams.processor.internals.DefaultStateUpdater$StateUpdaterThread.resumeTasks(DefaultStateUpdater.java:241)
> at
> org.apache.kafka.streams.processor.internals.DefaultStateUpdater$StateUpdaterThread.runOnce(DefaultStateUpdater.java:197)
> at
> org.apache.kafka.streams.processor.internals.DefaultStateUpdater$StateUpdaterThread.run(DefaultStateUpdater.java:163)
> transitToUpdateStandby() is documented as not idempotent and throws when the
> reader state is not ACTIVE_RESTORING. Three call sites on the state updater
> thread call it without checking the state first:
> 1. addTask, standby branch, when updatingTasks.size() == 1.
> 2. resumeTask, standby branch, when updatingTasks.size() == 1.
> 3. transitToUpdateStandbysIfOnlyStandbysLeft, when onlyStandbyTasksUpdating()
> is true.
> This is a single thread problem and needs no cross thread race. Ordering on
> the state updater thread:
> -> the reader is already STANDBY_UPDATING, which is the normal state once
> only standbys are left,
> -> the updating tasks are paused (pauseTask does not move the reader back to
> ACTIVE_RESTORING),
> -> one standby is added or resumed, updatingTasks.size() becomes 1, the call
> to transitToUpdateStandby() runs on a reader that is already
> STANDBY_UPDATING, and it throws, killing the state updater thread.
> Reproduced on 4.1.2 and 4.3.0 with a standalone Streams application under
> topology pause and resume churn (the same reproducer used for KAFKA-20724).
> The code path is unchanged on trunk.
> Relationship to KAFKA-17946: that ticket was the flaky test
> shouldResumeStandbyTask. Its fix (PR #18253) stabilized the test but did not
> change resumeTask or transitToUpdateStandby, so the production transition is
> still not idempotent and still reachable.
> Suggested fix: preserve the invariant that transitToUpdateStandby is only
> called when the reader is ACTIVE_RESTORING. Guard all three call sites with
> isRestoringActive(), or make transitToUpdateStandby idempotent so it is a no
> op when the reader is already STANDBY_UPDATING, which covers every caller in
> one place.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
