cadonna commented on code in PR #12200:
URL: https://github.com/apache/kafka/pull/12200#discussion_r880396183
##########
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);
} finally {
clear();
+ shutdownGate.countDown();
+ log.info("State updater thread shutdown");
}
}
- private void performActionsOnTasks() throws InterruptedException {
+ private void runOnce() throws InterruptedException {
+ performActionsOnTasks();
+ restoreTasks();
+ waitIfAllChangelogsCompletelyRead();
+ }
+
+ private void performActionsOnTasks() {
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
