[
https://issues.apache.org/jira/browse/KAFKA-20731?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18134090#comment-18134090
]
Eswarar Siva commented on KAFKA-20731:
--------------------------------------
Pause hits a related case with a different exception: the switch to
STANDBY_UPDATING happens while a paused active changelog is still RESTORING.
Seen on 4.3.2 in a deployment with exactly_once_v2 and num.standby.replicas=0
(the standbys were warm-up replicas), then reproduced with two instances, an
active placed to be restoring and a standby updating on the same state updater,
and one KafkaStreams#pause():
{noformat}
java.lang.IllegalStateException: Should not be in standby updating state if
there are still un-completed active changelogs
at
org.apache.kafka.streams.processor.internals.StoreChangelogReader.restore(StoreChangelogReader.java:482)
at
org.apache.kafka.streams.processor.internals.DefaultStateUpdater$StateUpdaterThread.restoreTasks(DefaultStateUpdater.java:273)
at
org.apache.kafka.streams.processor.internals.DefaultStateUpdater$StateUpdaterThread.runOnce(DefaultStateUpdater.java:213)
at
org.apache.kafka.streams.processor.internals.DefaultStateUpdater$StateUpdaterThread.run(DefaultStateUpdater.java:166)
{noformat}
pauseTasks() pauses the tasks in the iteration order of updatingTasks. When the
last restoring active is paused while a standby is still updating, pauseTask()
calls transitToUpdateStandbysIfOnlyStandbysLeft() and the reader goes to
STANDBY_UPDATING. The paused active keeps its changelog registered and
RESTORING, so the next restore() throws. The stream thread then fails with "The
state updater died and cannot update tasks anymore", its tasks close dirty and
under EOS their state directories are wiped, so the restore starts again from
the first offset.
Pause order as logged by the state updater:
* active restoring, standby: dies
* two actives restoring, standby: dies
* standby, active restoring, standby (the deployment's order): dies
* active restoring alone, two actives restoring, standby then active,
active/standby/active: no error
pauseTask() is the same on 4.4.0 and trunk (DefaultStateUpdater 725-727,
StoreChangelogReader 497-498 on trunk). Reading PR #22864, it covers this
order: on the last pauseTask() updatingTasks is empty, so
maybeTransitionChangelogReader() calls enforceRestoreActive() before the next
restore(). I have not run the PR. DefaultStateUpdaterTest mocks the
ChangelogReader, so no test there reaches the restore() check for this order. I
can share the reproduction if that helps.
> 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
> Assignee: Arnab Nandy
> 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
(v10.3.26#10030026)
