cadonna commented on code in PR #12270:
URL: https://github.com/apache/kafka/pull/12270#discussion_r892860698
##########
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() {
Review Comment:
Added a couple of methods `get*` to improve unit testing. Those methods
allow to look into the queues without draining them. They are not part of the
`StateUpdater` interface.
--
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]
