[GitHub] [kafka] cadonna commented on a diff in pull request #12200: KAFKA-10199: Implement adding standby tasks to the state updater

[GitBox <gi...@apache.org>]

cadonna commented on code in PR #12200:
URL: https://github.com/apache/kafka/pull/12200#discussion_r880392811


##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java:
##########
@@ -74,30 +76,44 @@ public Collection<Task> getAllUpdatingTasks() {
             return updatingTasks.values();
         }
 
+        public Collection<StandbyTask> getUpdatingStandbyTasks() {
+            return updatingTasks.values().stream()
+                .filter(t -> !t.isActive())
+                .map(t -> (StandbyTask) t)
+                .collect(Collectors.toList());
+        }
+
+        public boolean onlyStandbyTasksLeft() {
+            return !updatingTasks.isEmpty() && updatingTasks.values().stream().allMatch(t -> !t.isActive());
+        }
+
         @Override
         public void run() {
+            log.info("State updater thread started");
             try {
                 while (isRunning.get()) {
                     try {
-                        performActionsOnTasks();
-                        restoreTasks();
-                        waitIfAllChangelogsCompletelyRead();
+                        runOnce();
                     } catch (final InterruptedException interruptedException) {
                         return;
                     }
                 }
             } catch (final RuntimeException anyOtherException) {
-                log.error("An unexpected error occurred within the state updater thread: " + anyOtherException);
-                final ExceptionAndTasks exceptionAndTasks = new ExceptionAndTasks(new HashSet<>(updatingTasks.values()), anyOtherException);
-                updatingTasks.clear();
-                failedTasks.add(exceptionAndTasks);
-                isRunning.set(false);
+                handleRuntimeException(anyOtherException);

Review Comment:
   Just refactoring.



-- 
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-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org