cadonna commented on code in PR #12442:
URL: https://github.com/apache/kafka/pull/12442#discussion_r931907686
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -380,6 +346,104 @@ public void handleAssignment(final Map<TaskId,
Set<TopicPartition>> activeTasks,
tasks.createTasks(activeTasksToCreate, standbyTasksToCreate);
}
+ private void classifyTasksWithoutStateUpdater(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()) {
+ 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 {
+ tasksToRecycle.put(task, activeTasksToCreate.get(taskId));
+ }
+ activeTasksToCreate.remove(taskId);
+ } else if (standbyTasksToCreate.containsKey(taskId)) {
+ if (!task.isActive()) {
+ final Set<TopicPartition> topicPartitions =
standbyTasksToCreate.get(taskId);
+ task.updateInputPartitions(topicPartitions,
topologyMetadata.nodeToSourceTopics(task.id()));
+ task.resume();
+ } else {
+ tasksToRecycle.put(task, standbyTasksToCreate.get(taskId));
+ }
+ standbyTasksToCreate.remove(taskId);
+ } else {
+ tasksToCloseClean.add(task);
+ }
+ }
+ }
+
+ 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) {
+ for (final Task task : tasks.allTasks()) {
+ 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");
+ }
+ 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));
+ }
+ standbyTasksToCreate.remove(taskId);
+ } else {
+ tasksToCloseClean.add(task);
+ }
+ }
+ }
+
+ private void classifyTasksWithStateUpdater(final Map<TaskId,
Set<TopicPartition>> activeTasksToCreate,
+ final Map<TaskId,
Set<TopicPartition>> standbyTasksToCreate,
+ final Map<Task,
Set<TopicPartition>> tasksToRecycle,
+ final Set<Task>
tasksToCloseClean) {
+ classifyRunningTasks(activeTasksToCreate, standbyTasksToCreate,
tasksToRecycle, tasksToCloseClean);
+ for (final Task task : stateUpdater.getTasks()) {
+ final TaskId taskId = task.id();
+ if (activeTasksToCreate.containsKey(taskId)) {
+ if (task.isActive()) {
+ final Set<TopicPartition> topicPartitions =
activeTasksToCreate.get(taskId);
+ if (!task.inputPartitions().equals(topicPartitions)) {
+
tasks.addPendingTaskThatNeedsInputPartitionsUpdate(taskId);
Review Comment:
I think such a comment will be out-dated quite quickly since when we finish
this work we will remove the old code path.
One idea that came to my mind now: Maybe we should provide an unmodifiable
version of active and standby tasks so that we can ensure that tasks that are
returned from `stateUpdater.getTasks()` cannot be modified. In such a way also
resuming such a task would throw an exception. I think that would be better
than a comment.
--
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
