cadonna commented on code in PR #12270:
URL: https://github.com/apache/kafka/pull/12270#discussion_r893836048
##########
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:
Since we will use the interface `StateUpdater` in production code and not
its implementation `DefaultStateUpdater` those getters should not be visible to
production code as long as no casting is used.
I am not a fan of the `for testing only` comment since it does not enforce
anything and only pollutes the code.
Moreover I think there are good chances that we will need those getters to
expose the tasks to the task manager. But let's see what the future will bring
us.
--
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]
