[
https://issues.apache.org/jira/browse/KAFKA-20827?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Alan Lau updated KAFKA-20827:
-----------------------------
Description:
With num.standby.replicas >= 1, a StreamThread can be killed by a fatal
IllegalStateException from the task registry while a task recycles between
active and standby:
{code:java}
java.lang.IllegalStateException: Attempted to create an standby task that we
already own: 0_2
at
org.apache.kafka.streams.processor.internals.Tasks.addStandbyTask(Tasks.java:221)
at
org.apache.kafka.streams.processor.internals.Tasks.addTask(Tasks.java:193)
at
org.apache.kafka.streams.processor.internals.Tasks.addFailedTask(Tasks.java:234)
at
org.apache.kafka.streams.processor.internals.TaskManager.addTasksToStateUpdater(TaskManager.java:944)
at
org.apache.kafka.streams.processor.internals.TaskManager.checkStateUpdater(TaskManager.java:855)
at
org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:1042)
{code}
A task whose initialization fails is left owned (registered in Tasks and
flagged in failedTaskIds) to be reconciled via the corruption/failed-task path.
handleAssignment's rectify-existing pass (handleTasks) iterates
allNonFailedInitializedTasks(), which excludes failed tasks so the
owned-but-failed task is not recognized as already-owned and is not removed
from the ...ToCreate maps. createNewTasks then builds a second representation
of it; that duplicate fails initialization again, and addTasksToStateUpdater ->
Tasks.addFailedTask -> addStandbyTask/addActiveTask trips the single-owner
invariant and kills the StreamThread. The throw also happens before the real
init exception is recorded, masking it.
was:
With num.standby.replicas >= 1, a StreamThread can be killed by a fatal
IllegalStateException from the task registry while a task recycles between
active and standby:
{code:java}
java.lang.IllegalStateException: Attempted to create an standby task that we
already own: 0_2
at
org.apache.kafka.streams.processor.internals.Tasks.addStandbyTask(Tasks.java:221)
at
org.apache.kafka.streams.processor.internals.Tasks.addTask(Tasks.java:193)
at
org.apache.kafka.streams.processor.internals.Tasks.addFailedTask(Tasks.java:234)
at
org.apache.kafka.streams.processor.internals.TaskManager.addTasksToStateUpdater(TaskManager.java:944)
at
org.apache.kafka.streams.processor.internals.TaskManager.checkStateUpdater(TaskManager.java:855)
at
org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:1042)
{code}
A recycle can leave a task double-represented:
TaskManager.recycleTaskFromStateUpdater and the assignment recycle suspend()
and convert the task, then enqueue the new representation via
tasks.addPendingTasksToInit(...), while the old representation is still
registered in activeTasksPerId/standbyTasksPerId. If the new representation
then fails to (re)initialize, TaskManager.addTasksToStateUpdater
(TaskManager.java:937-950) catches the RuntimeException and calls
Tasks.addFailedTask, which unconditionally calls addTask then
addStandbyTask/addActiveTask and trips the single-owner invariant, killing the
thread.
> StreamThread dies with IllegalStateException: Attempted to create an standby
> task that we already own during a standby/active recycle
> -------------------------------------------------------------------------------------------------------------------------------------
>
> Key: KAFKA-20827
> URL: https://issues.apache.org/jira/browse/KAFKA-20827
> Project: Kafka
> Issue Type: Bug
> Components: streams
> Reporter: Alan Lau
> Assignee: Alan Lau
> Priority: Major
>
> With num.standby.replicas >= 1, a StreamThread can be killed by a fatal
> IllegalStateException from the task registry while a task recycles between
> active and standby:
> {code:java}
> java.lang.IllegalStateException: Attempted to create an standby task that we
> already own: 0_2
>
> at
> org.apache.kafka.streams.processor.internals.Tasks.addStandbyTask(Tasks.java:221)
>
>
> at
> org.apache.kafka.streams.processor.internals.Tasks.addTask(Tasks.java:193)
>
>
> at
> org.apache.kafka.streams.processor.internals.Tasks.addFailedTask(Tasks.java:234)
>
>
> at
> org.apache.kafka.streams.processor.internals.TaskManager.addTasksToStateUpdater(TaskManager.java:944)
>
>
> at
> org.apache.kafka.streams.processor.internals.TaskManager.checkStateUpdater(TaskManager.java:855)
>
>
> at
> org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:1042)
>
> {code}
> A task whose initialization fails is left owned (registered in Tasks and
> flagged in failedTaskIds) to be reconciled via the corruption/failed-task
> path.
> handleAssignment's rectify-existing pass (handleTasks) iterates
> allNonFailedInitializedTasks(), which excludes failed tasks so the
> owned-but-failed task is not recognized as already-owned and is not removed
> from the ...ToCreate maps. createNewTasks then builds a second representation
> of it; that duplicate fails initialization again, and addTasksToStateUpdater
> -> Tasks.addFailedTask -> addStandbyTask/addActiveTask trips the single-owner
> invariant and kills the StreamThread. The throw also happens before the real
> init exception is recorded, masking it.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
