[
https://issues.apache.org/jira/browse/KAFKA-20731?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18090918#comment-18090918
]
Eswarar Siva commented on KAFKA-20731:
--------------------------------------
[~lucasbru] I will take the PR for this. There are two ways to enforce the
invariant and I would like your preference before I pick one.
Option 1=> guard the callers. transitToUpdateStandby is reached from three
places on the state updater thread, addTask and resumeTask in their standby
branch when updatingTasks.size() == 1 and
transitToUpdateStandbysIfOnlyStandbysLeft. I would add an isRestoringActive()
check to each so the
call only happens when the reader is ACTIVE_RESTORING. This matches the
invariant you described and keeps the existing throw as a guard against a
future caller that gets it wrong. It does touch a number of existing
DefaultStateUpdaterTest cases, since they mock the reader and would need
isRestoringActive() to return true in the standby transition flows.
Option 2: make transitToUpdateStandby idempotent, so it returns early when the
reader is already STANDBY_UPDATING instead of throwing. Smaller change and it
covers every caller in one place, but it loosens the current "not idempotent"
contract and drops the throw that catches an unexpected call.
I lean towards option 1 since it keeps the throw and matches what you said
about only transitioning from ACTIVE_RESTORING, but option 2 is less code if
you would rather keep it minimal. Which do you prefer?
> 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)
