guozhangwang commented on code in PR #12466:
URL: https://github.com/apache/kafka/pull/12466#discussion_r946143073
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -656,6 +657,71 @@ boolean tryToCompleteRestoration(final long now, final
java.util.function.Consum
return allRunning;
}
+ private void addTaskstoStateUpdater() {
+ for (final Task task : tasks.drainPendingTaskToInit()) {
+ task.initializeIfNeeded();
+ stateUpdater.add(task);
+ }
+ }
+
+ private void handleRemovedTasksFromStateUpdater() {
+ final Map<TaskId, RuntimeException> taskExceptions = new
LinkedHashMap<>();
+ final Set<Task> tasksToCloseDirty = new
TreeSet<>(Comparator.comparing(Task::id));
+
+ for (final Task task : stateUpdater.drainRemovedTasks()) {
+ final TaskId taskId = task.id();
+ Set<TopicPartition> inputPartitions;
+ if ((inputPartitions =
tasks.removePendingTaskToRecycle(task.id())) != null) {
+ try {
+ final Task newTask = task.isActive() ?
+ convertActiveToStandby((StreamTask) task,
inputPartitions) :
+ convertStandbyToActive((StandbyTask) task,
inputPartitions);
+ newTask.initializeIfNeeded();
+ stateUpdater.add(newTask);
+ } catch (final RuntimeException e) {
+ final String uncleanMessage = String.format("Failed to
recycle task %s cleanly. " +
+ "Attempting to close remaining tasks before
re-throwing:", taskId);
+ log.error(uncleanMessage, e);
+
+ if (task.state() != State.CLOSED) {
+ tasksToCloseDirty.add(task);
+ }
+
+ taskExceptions.putIfAbsent(taskId, e);
+ }
+ } else if (tasks.removePendingTaskToClose(task.id())) {
+ try {
+ task.closeClean();
Review Comment:
You're right! Added and also updated unit tests.
--
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
