cadonna commented on code in PR #12600:
URL: https://github.com/apache/kafka/pull/12600#discussion_r966771190
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -421,73 +421,120 @@ private void classifyTasksWithoutStateUpdater(final
Map<TaskId, Set<TopicPartiti
}
}
- private void classifyRunningTasks(final Map<TaskId, Set<TopicPartition>>
activeTasksToCreate,
- final Map<TaskId, Set<TopicPartition>>
standbyTasksToCreate,
- final Map<Task, Set<TopicPartition>>
tasksToRecycle,
- final Set<Task> tasksToCloseClean) {
+ private void classifyRunningAndSuspendedTasks(final Map<TaskId,
Set<TopicPartition>> activeTasksToCreate,
+ final Map<TaskId,
Set<TopicPartition>> standbyTasksToCreate,
+ final Map<Task,
Set<TopicPartition>> tasksToRecycle,
+ final Set<Task>
tasksToCloseClean) {
for (final Task task : tasks.allTasks()) {
+ if (!task.isActive()) {
+ throw new IllegalStateException("Standby tasks should only be
managed by the state updater");
+ }
final TaskId taskId = task.id();
if (activeTasksToCreate.containsKey(taskId)) {
- if (task.isActive()) {
- final Set<TopicPartition> topicPartitions =
activeTasksToCreate.get(taskId);
- if (tasks.updateActiveTaskInputPartitions(task,
topicPartitions)) {
- task.updateInputPartitions(topicPartitions,
topologyMetadata.nodeToSourceTopics(task.id()));
- }
- task.resume();
- } else {
- throw new IllegalStateException("Standby tasks should only
be managed by the state updater");
- }
+ handleReAssignedActiveTask(task,
activeTasksToCreate.get(taskId));
activeTasksToCreate.remove(taskId);
} else if (standbyTasksToCreate.containsKey(taskId)) {
- if (!task.isActive()) {
- throw new IllegalStateException("Standby tasks should only
be managed by the state updater");
- } else {
- tasksToRecycle.put(task, standbyTasksToCreate.get(taskId));
- }
+ tasksToRecycle.put(task, standbyTasksToCreate.get(taskId));
standbyTasksToCreate.remove(taskId);
} else {
tasksToCloseClean.add(task);
}
}
}
+ private void handleReAssignedActiveTask(final Task task,
+ final Set<TopicPartition>
inputPartitions) {
+ if (tasks.updateActiveTaskInputPartitions(task, inputPartitions)) {
+ task.updateInputPartitions(inputPartitions,
topologyMetadata.nodeToSourceTopics(task.id()));
+ }
+ task.resume();
Review Comment:
Method ` handleReAssignedActiveTask()` is called in
`classifyRunningAndSuspendedTasks()`. Method
`classifyRunningAndSuspendedTasks()` classifies the tasks that are `RUNNING` or
`SUSPENDED`. In other words, the tasks that are in the tasks registry. Those
tasks are not in state `RESTORING` and so the `if`-branch in your proposal
would never be executed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
