MINOR: rename TaskRegistry methods to better reflect their purpose. (apache#21448)

Changed the name of method that work only with initialized tasks(not
pending) to better reflect their purpose.

Reviewers: Matthias J. Sax <matthias@confluent.io>, Lucas Brutschy
 <lbrutschy@confluent.io>

Author: Nikita-Shupletsov

streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskExecutor.java

@@ -68,7 +68,7 @@ int process(final int maxNumRecords, final Time time) {
         int totalProcessed = 0;
         Task lastProcessed = null;
 
-        for (final Task task : tasks.activeTasks()) {
+        for (final Task task : tasks.activeInitializedTasks()) {
             final long now = time.milliseconds();
             try {
                 if (executionMetadata.canProcessTask(task, now)) {
@@ -233,7 +233,7 @@ void commitOffsetsOrTransaction(final Map<Task, Map<TopicPartition, OffsetAndMet
 
     private void updateTaskCommitMetadata(final Map<TopicPartition, OffsetAndMetadata> allOffsets) {
         if (!allOffsets.isEmpty()) {
-            for (final Task task : tasks.activeTasks()) {
+            for (final Task task : tasks.activeInitializedTasks()) {
                 if (task instanceof StreamTask) {
                     for (final TopicPartition topicPartition : task.inputPartitions()) {
                         if (allOffsets.containsKey(topicPartition)) {
@@ -261,7 +261,7 @@ private void commitSuccessfullyProcessedTasks() {
     int punctuate() {
         int punctuated = 0;
 
-        for (final Task task : tasks.activeTasks()) {
+        for (final Task task : tasks.activeInitializedTasks()) {
             try {
                 if (executionMetadata.canPunctuateTask(task)) {
                     if (task.maybePunctuateStreamTime()) {

streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java

@@ -197,7 +197,7 @@ void handleRebalanceComplete() {
         // we should pause consumer only within the listener since
         // before then the assignment has not been updated yet.
         // All tasks that are owned by the task manager are ready and do not need to be paused
-        final Set<TopicPartition> partitionsNotToPause = tasks.allNonFailedTasks()
+        final Set<TopicPartition> partitionsNotToPause = tasks.allNonFailedInitializedTasks()
             .stream()
             .flatMap(task -> task.inputPartitions().stream())
             .collect(Collectors.toSet());
@@ -214,7 +214,7 @@ void handleRebalanceComplete() {
      * @throws TaskMigratedException
      */
     boolean handleCorruption(final Set<TaskId> corruptedTasks) {
-        final Set<TaskId> activeTasks = new HashSet<>(tasks.activeTaskIds());
+        final Set<TaskId> activeTasks = new HashSet<>(tasks.activeInitializedTaskIds());
 
         // We need to stop all processing, since we need to commit non-corrupted tasks as well.
         maybeLockTasks(activeTasks);
@@ -223,7 +223,7 @@ boolean handleCorruption(final Set<TaskId> corruptedTasks) {
         final Set<Task> corruptedStandbyTasks = new TreeSet<>(Comparator.comparing(Task::id));
 
         for (final TaskId taskId : corruptedTasks) {
-            final Task task = tasks.task(taskId);
+            final Task task = tasks.initializedTask(taskId);
             if (task.isActive()) {
                 corruptedActiveTasks.add(task);
             } else {
@@ -237,7 +237,7 @@ boolean handleCorruption(final Set<TaskId> corruptedTasks) {
 
         // We need to commit before closing the corrupted active tasks since this will force the ongoing txn to abort
         try {
-            final Collection<Task> tasksToCommit = tasks.allTasksPerId()
+            final Collection<Task> tasksToCommit = tasks.allInitializedTasksPerId()
                 .values()
                 .stream()
                 .filter(t -> t.state() == Task.State.RUNNING)
@@ -247,10 +247,10 @@ boolean handleCorruption(final Set<TaskId> corruptedTasks) {
         } catch (final TaskCorruptedException e) {
             log.info("Some additional tasks were found corrupted while trying to commit, these will be added to the " +
                          "tasks to clean and revive: {}", e.corruptedTasks());
-            corruptedActiveTasks.addAll(tasks.tasks(e.corruptedTasks()));
+            corruptedActiveTasks.addAll(tasks.initializedTasks(e.corruptedTasks()));
         } catch (final TimeoutException e) {
             log.info("Hit TimeoutException when committing all non-corrupted tasks, these will be closed and revived");
-            final Collection<Task> uncorruptedTasks = new HashSet<>(tasks.activeTasks());
+            final Collection<Task> uncorruptedTasks = new HashSet<>(tasks.activeInitializedTasks());
             uncorruptedTasks.removeAll(corruptedActiveTasks);
             // Those tasks which just timed out can just be closed dirty without marking changelogs as corrupted
             closeDirtyAndRevive(uncorruptedTasks, false);
@@ -366,7 +366,7 @@ public void handleAssignment(final Map<TaskId, Set<TopicPartition>> activeTasks,
         final Set<Task> tasksToCloseClean = new TreeSet<>(Comparator.comparing(Task::id));
 
         final Set<TaskId> tasksToLock =
-            tasks.allTaskIds().stream()
+            tasks.allInitializedTaskIds().stream()
                 .filter(x -> activeTasksToCreate.containsKey(x) || standbyTasksToCreate.containsKey(x))
                 .collect(Collectors.toSet());
 
@@ -533,7 +533,7 @@ private void handleRunningAndSuspendedTasks(final Map<TaskId, Set<TopicPartition
                                                 final Map<TaskId, Set<TopicPartition>> standbyTasksToCreate,
                                                 final Map<Task, Set<TopicPartition>> tasksToRecycle,
                                                 final Set<Task> tasksToCloseClean) {
-        for (final Task task : tasks.allNonFailedTasks()) {
+        for (final Task task : tasks.allNonFailedInitializedTasks()) {
             if (!task.isActive()) {
                 throw new IllegalStateException("Standby tasks should only be managed by the state updater, " +
                     "but standby task " + task.id() + " is managed by the stream thread");
@@ -733,7 +733,7 @@ private Map<TaskId, Set<TopicPartition>> pendingTasksToCreate(final Map<TaskId,
         while (iter.hasNext()) {
             final Map.Entry<TaskId, Set<TopicPartition>> entry = iter.next();
             final TaskId taskId = entry.getKey();
-            final boolean taskIsOwned = tasks.allTaskIds().contains(taskId)
+            final boolean taskIsOwned = tasks.allInitializedTaskIds().contains(taskId)
                 || (stateUpdater.tasks().stream().anyMatch(task -> task.id().equals(taskId)));
             if (taskId.topologyName() != null && !taskIsOwned && !topologyMetadata.namedTopologiesView().contains(taskId.topologyName())) {
                 log.info("Cannot create the assigned task {} since it's topology name cannot be recognized, will put it " +
@@ -1079,7 +1079,7 @@ void handleRevocation(final Collection<TopicPartition> revokedPartitions) {
                      e.corruptedTasks());
 
             // If we hit a TaskCorruptedException it must be EOS, just handle the cleanup for those corrupted tasks right here
-            dirtyTasks.addAll(tasks.tasks(e.corruptedTasks()));
+            dirtyTasks.addAll(tasks.initializedTasks(e.corruptedTasks()));
             closeDirtyAndRevive(dirtyTasks, true);
         } catch (final TimeoutException e) {
             log.warn("Timed out while trying to commit all tasks during revocation, these will be cleaned and revived");
@@ -1193,8 +1193,8 @@ void handleLostAll() {
     }
 
     private void closeRunningTasksDirty() {
-        final Set<Task> allTask = tasks.allTasks();
-        final Set<TaskId> allTaskIds = tasks.allTaskIds();
+        final Set<Task> allTask = tasks.allInitializedTasks();
+        final Set<TaskId> allTaskIds = tasks.allInitializedTaskIds();
         maybeLockTasks(allTaskIds);
         for (final Task task : allTask) {
             // Even though we've apparently dropped out of the group, we can continue safely to maintain our
@@ -1413,10 +1413,10 @@ void shutdown(final boolean clean) {
 
         // TODO: change type to `StreamTask`
         final Set<Task> activeTasks = new TreeSet<>(Comparator.comparing(Task::id));
-        activeTasks.addAll(tasks.activeTasks());
+        activeTasks.addAll(tasks.activeInitializedTasks());
         // TODO: change type to `StandbyTask`
         final Set<Task> standbyTasks = new TreeSet<>(Comparator.comparing(Task::id));
-        standbyTasks.addAll(tasks.standbyTasks());
+        standbyTasks.addAll(tasks.standbyInitializedTasks());
 
         final Set<Task> pendingActiveTasks = tasks.drainPendingActiveTasksToInit();
         activeTasks.addAll(pendingActiveTasks);
@@ -1670,7 +1670,7 @@ Map<TaskId, Task> allTasks() {
         // not bothering with an unmodifiable map, since the tasks themselves are mutable, but
         // if any outside code modifies the map or the tasks, it would be a severe transgression.
         final Map<TaskId, Task> ret = stateUpdater.tasks().stream().collect(Collectors.toMap(Task::id, x -> x));
-        ret.putAll(tasks.allTasksPerId());
+        ret.putAll(tasks.allInitializedTasksPerId());
         ret.putAll(tasks.pendingTasksToInit().stream().collect(Collectors.toMap(Task::id, x -> x)));
         return ret;
     }
@@ -1683,19 +1683,19 @@ Map<TaskId, Task> allTasks() {
     Map<TaskId, Task> allRunningTasks() {
         // not bothering with an unmodifiable map, since the tasks themselves are mutable, but
         // if any outside code modifies the map or the tasks, it would be a severe transgression.
-        return tasks.allTasksPerId();
+        return tasks.allInitializedTasksPerId();
     }
 
     Set<Task> readOnlyAllTasks() {
         // not bothering with an unmodifiable map, since the tasks themselves are mutable, but
         // if any outside code modifies the map or the tasks, it would be a severe transgression.
         final HashSet<Task> ret = new HashSet<>(stateUpdater.tasks());
-        ret.addAll(tasks.allTasks());
+        ret.addAll(tasks.allInitializedTasks());
         return Collections.unmodifiableSet(ret);
     }
 
     Map<TaskId, Task> notPausedTasks() {
-        return Collections.unmodifiableMap(tasks.allTasks()
+        return Collections.unmodifiableMap(tasks.allInitializedTasks()
             .stream()
             .filter(t -> !topologyMetadata.isPaused(t.id().topologyName()))
             .collect(Collectors.toMap(Task::id, v -> v)));
@@ -1721,7 +1721,7 @@ private Stream<Task> activeTaskStream() {
     }
 
     private Stream<Task> activeRunningTaskStream() {
-        return tasks.allTasks().stream().filter(Task::isActive);
+        return tasks.allInitializedTasks().stream().filter(Task::isActive);
     }
 
     Map<TaskId, Task> standbyTaskMap() {
@@ -1733,23 +1733,23 @@ private List<Task> standbyTaskIterable() {
     }
 
     private Stream<Task> standbyTaskStream() {
-        final Stream<Task> standbyTasksInTaskRegistry = tasks.allTasks().stream().filter(t -> !t.isActive());
+        final Stream<Task> standbyTasksInTaskRegistry = tasks.allInitializedTasks().stream().filter(t -> !t.isActive());
         return Stream.concat(
             stateUpdater.standbyTasks().stream(),
             standbyTasksInTaskRegistry
         );
     }
     // For testing only.
     int commitAll() {
-        return commit(tasks.allTasks());
+        return commit(tasks.allInitializedTasks());
     }
 
     /**
      * Resumes polling in the main consumer for all partitions for which
      * the corresponding record queues have capacity (again).
      */
     public void resumePollingForPartitionsWithAvailableSpace() {
-        for (final Task t: tasks.activeTasks()) {
+        for (final Task t: tasks.activeInitializedTasks()) {
             t.resumePollingForPartitionsWithAvailableSpace();
         }
     }
@@ -1758,7 +1758,7 @@ public void resumePollingForPartitionsWithAvailableSpace() {
      * Fetches up-to-date lag information from the consumer.
      */
     public void updateLags() {
-        for (final Task t: tasks.activeTasks()) {
+        for (final Task t: tasks.activeInitializedTasks()) {
             t.updateLags();
         }
     }
@@ -1808,7 +1808,7 @@ void maybeInitTaskTimeoutsOrThrow(
     }
 
     private Task getActiveTask(final TopicPartition partition) {
-        final Task activeTask = tasks.activeTasksForInputPartition(partition);
+        final Task activeTask = tasks.activeInitializedTasksForInputPartition(partition);
 
         if (activeTask == null) {
             log.error("Unable to locate active task for received-record partition {}. Current tasks: {}",
@@ -1912,7 +1912,7 @@ private int commitTasksAndMaybeUpdateCommittableOffsets(final Collection<Task> t
     }
 
     public void updateTaskEndMetadata(final TopicPartition topicPartition, final Long offset) {
-        for (final Task task : tasks.activeTasks()) {
+        for (final Task task : tasks.activeInitializedTasks()) {
             if (task instanceof StreamTask) {
                 if (task.inputPartitions().contains(topicPartition)) {
                     ((StreamTask) task).updateEndOffsets(topicPartition, offset);
@@ -1943,7 +1943,7 @@ void maybeCloseTasksFromRemovedTopologies(final Set<String> currentNamedTopologi
         try {
             final Set<Task> activeTasksToRemove = new TreeSet<>(Comparator.comparing(Task::id));
             final Set<Task> standbyTasksToRemove = new TreeSet<>(Comparator.comparing(Task::id));
-            for (final Task task : tasks.allTasks()) {
+            for (final Task task : tasks.allInitializedTasks()) {
                 if (!currentNamedTopologies.contains(task.id().topologyName())) {
                     if (task.isActive()) {
                         activeTasksToRemove.add(task);
@@ -2032,7 +2032,7 @@ public String toString(final String indent) {
         stringBuilder.append("TaskManager\n");
         stringBuilder.append(indent).append("\tMetadataState:\n");
         stringBuilder.append(indent).append("\tTasks:\n");
-        for (final Task task : tasks.allTasks()) {
+        for (final Task task : tasks.allInitializedTasks()) {
             stringBuilder.append(indent)
                          .append("\t\t")
                          .append(task.id())

streams/src/main/java/org/apache/kafka/streams/processor/internals/Tasks.java

@@ -280,6 +280,7 @@ private void removePartitionsForActiveTask(final TaskId taskId) {
     @Override
     public synchronized void clear() {
         pendingTasksToInit.clear();
+        pendingTasksToClose.clear();
         pendingActiveTasksToCreate.clear();
         pendingStandbyTasksToCreate.clear();
         activeTasksPerId.clear();
@@ -290,7 +291,7 @@ public synchronized void clear() {
 
     // TODO: change return type to `StreamTask`
     @Override
-    public Task activeTasksForInputPartition(final TopicPartition partition) {
+    public Task activeInitializedTasksForInputPartition(final TopicPartition partition) {
         return activeTasksPerPartition.get(partition);
     }
 
@@ -305,7 +306,7 @@ private synchronized Task getTask(final TaskId taskId) {
     }
 
     @Override
-    public Task task(final TaskId taskId) {
+    public Task initializedTask(final TaskId taskId) {
         final Task task = getTask(taskId);
 
         if (task != null)
@@ -315,26 +316,26 @@ public Task task(final TaskId taskId) {
     }
 
     @Override
-    public Collection<Task> tasks(final Collection<TaskId> taskIds) {
+    public Collection<Task> initializedTasks(final Collection<TaskId> taskIds) {
         final Set<Task> tasks = new HashSet<>();
         for (final TaskId taskId : taskIds) {
-            tasks.add(task(taskId));
+            tasks.add(initializedTask(taskId));
         }
         return tasks;
     }
 
     @Override
-    public synchronized Collection<TaskId> activeTaskIds() {
+    public synchronized Collection<TaskId> activeInitializedTaskIds() {
         return Collections.unmodifiableCollection(activeTasksPerId.keySet());
     }
 
     @Override
-    public synchronized Collection<Task> activeTasks() {
+    public synchronized Collection<Task> activeInitializedTasks() {
         return Collections.unmodifiableCollection(activeTasksPerId.values());
     }
 
     @Override
-    public synchronized Collection<Task> standbyTasks() {
+    public synchronized Collection<Task> standbyInitializedTasks() {
         return Collections.unmodifiableCollection(standbyTasksPerId.values());
     }
 
@@ -343,12 +344,12 @@ public synchronized Collection<Task> standbyTasks() {
      * and the returned task could be modified by other threads concurrently
      */
     @Override
-    public synchronized Set<Task> allTasks() {
+    public synchronized Set<Task> allInitializedTasks() {
         return union(HashSet::new, new HashSet<>(activeTasksPerId.values()), new HashSet<>(standbyTasksPerId.values()));
     }
 
     @Override
-    public synchronized Set<Task> allNonFailedTasks() {
+    public synchronized Set<Task> allNonFailedInitializedTasks() {
         final Set<Task> nonFailedActiveTasks = activeTasksPerId.values().stream()
             .filter(task -> !failedTaskIds.contains(task.id()))
             .collect(Collectors.toSet());
@@ -359,20 +360,20 @@ public synchronized Set<Task> allNonFailedTasks() {
     }
 
     @Override
-    public synchronized Set<TaskId> allTaskIds() {
+    public synchronized Set<TaskId> allInitializedTaskIds() {
         return union(HashSet::new, activeTasksPerId.keySet(), standbyTasksPerId.keySet());
     }
 
     @Override
-    public synchronized Map<TaskId, Task> allTasksPerId() {
+    public synchronized Map<TaskId, Task> allInitializedTasksPerId() {
         final Map<TaskId, Task> ret = new HashMap<>();
         ret.putAll(activeTasksPerId);
         ret.putAll(standbyTasksPerId);
         return ret;
     }
 
     @Override
-    public boolean contains(final TaskId taskId) {
+    public boolean containsInitialized(final TaskId taskId) {
         return getTask(taskId) != null;
     }
 

streams/src/main/java/org/apache/kafka/streams/processor/internals/TasksRegistry.java

@@ -69,25 +69,25 @@ public interface TasksRegistry {
 
     void clear();
 
-    Task activeTasksForInputPartition(final TopicPartition partition);
+    Task activeInitializedTasksForInputPartition(final TopicPartition partition);
 
-    Task task(final TaskId taskId);
+    Task initializedTask(final TaskId taskId);
 
-    Collection<Task> tasks(final Collection<TaskId> taskIds);
+    Collection<Task> initializedTasks(final Collection<TaskId> taskIds);
 
-    Collection<TaskId> activeTaskIds();
+    Collection<TaskId> activeInitializedTaskIds();
 
-    Collection<Task> activeTasks();
+    Collection<Task> activeInitializedTasks();
 
-    Collection<Task> standbyTasks();
+    Collection<Task> standbyInitializedTasks();
 
-    Set<Task> allTasks();
+    Set<Task> allInitializedTasks();
 
-    Set<Task> allNonFailedTasks();
+    Set<Task> allNonFailedInitializedTasks();
 
-    Map<TaskId, Task> allTasksPerId();
+    Map<TaskId, Task> allInitializedTasksPerId();
 
-    Set<TaskId> allTaskIds();
+    Set<TaskId> allInitializedTaskIds();
 
-    boolean contains(final TaskId taskId);
+    boolean containsInitialized(final TaskId taskId);
 }