guozhangwang commented on code in PR #12270:
URL: https://github.com/apache/kafka/pull/12270#discussion_r893842474
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java:
##########
@@ -375,52 +384,42 @@ public Set<StreamTask> getRestoredActiveTasks(final
Duration timeout) {
}
@Override
- public List<ExceptionAndTasks> getFailedTasksAndExceptions() {
+ public Set<Task> drainRemovedTasks() {
+ final List<Task> result = new ArrayList<>();
+ removedTasks.drainTo(result);
+ return new HashSet<>(result);
+ }
+
+ @Override
+ public List<ExceptionAndTasks> drainExceptionsAndFailedTasks() {
final List<ExceptionAndTasks> result = new ArrayList<>();
- failedTasks.drainTo(result);
+ exceptionsAndFailedTasks.drainTo(result);
return result;
}
- @Override
- public Set<Task> getAllTasks() {
- tasksAndActionsLock.lock();
+ public Set<StandbyTask> getUpdatingStandbyTasks() {
+ return Collections.unmodifiableSet(new
HashSet<>(stateUpdaterThread.getUpdatingStandbyTasks()));
+ }
+
+ public Set<Task> getUpdatingTasks() {
+ return Collections.unmodifiableSet(new
HashSet<>(stateUpdaterThread.getUpdatingTasks()));
+ }
+
+ public Set<StreamTask> getRestoredActiveTasks() {
restoredActiveTasksLock.lock();
try {
- final Set<Task> allTasks = new HashSet<>();
- allTasks.addAll(tasksAndActions.stream()
- .filter(t -> t.action == Action.ADD)
- .map(t -> t.task)
- .collect(Collectors.toList())
- );
- allTasks.addAll(stateUpdaterThread.getAllUpdatingTasks());
- allTasks.addAll(restoredActiveTasks);
- return Collections.unmodifiableSet(allTasks);
+ return Collections.unmodifiableSet(new
HashSet<>(restoredActiveTasks));
} finally {
restoredActiveTasksLock.unlock();
- tasksAndActionsLock.unlock();
}
}
- @Override
- public Set<StandbyTask> getStandbyTasks() {
- tasksAndActionsLock.lock();
- try {
- final Set<StandbyTask> standbyTasks = new HashSet<>();
- standbyTasks.addAll(tasksAndActions.stream()
- .filter(t -> t.action == Action.ADD)
- .filter(t -> !t.task.isActive())
- .map(t -> (StandbyTask) t.task)
- .collect(Collectors.toList())
- );
- standbyTasks.addAll(getUpdatingStandbyTasks());
- return Collections.unmodifiableSet(standbyTasks);
- } finally {
- tasksAndActionsLock.unlock();
- }
+ public List<ExceptionAndTasks> getExceptionsAndFailedTasks() {
Review Comment:
Sounds good. Not advocating for the `for testing only comment` here, maybe
just my paranoid gene that would like to see those functions clustered together
but not feeling strong either way :)
--
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]
