[
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.
The StreamThread should keep running and only the affected task briefly
re-restores before resuming, with no impact on the other tasks on that thread.
Instead, it throws a IllegalStateException that kills the entire StreamThread.
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 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.
> 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.
> The StreamThread should keep running and only the affected task briefly
> re-restores before resuming, with no impact on the other tasks on that
> thread. Instead, it throws a IllegalStateException that kills the entire
> StreamThread.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
