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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
