KAFKA-19510: clear pendingTasksToInit on tasks clear. (#20646)

Clear pendingTasksToInit on tasks clear.  It matters in situations when
we shutting down a thread in PARTITIONS_ASSIGNED state. In this case we
may have locked some unassigned task directories (see
TaskManager#tryToLockAllNonEmptyTaskDirectories). Then we may have
gotten assigned to one or multiple of those tasks. In this scenario,  we
will not release the locks for the unassigned task directories (see
TaskManager#releaseLockedUnassignedTaskDirectories), because
TaskManager#allTasks includes pendingTasksToInit, but it hasn't been
cleared.

Reviewers: Matthias J. Sax <[email protected]>, Lucas Brutschy
 <[email protected]>

Author: Nikita-Shupletsov

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

@@ -258,6 +258,9 @@ private void removePartitionsForActiveTask(final TaskId taskId) {
 
     @Override
     public synchronized void clear() {
+        pendingTasksToInit.clear();
+        pendingActiveTasksToCreate.clear();
+        pendingStandbyTasksToCreate.clear();
         activeTasksPerId.clear();
         standbyTasksPerId.clear();
         activeTasksPerPartition.clear();
@@ -346,4 +349,12 @@ public synchronized Map<TaskId, Task> allTasksPerId() {
     public boolean contains(final TaskId taskId) {
         return getTask(taskId) != null;
     }
+
+    Map<TaskId, Set<TopicPartition>> pendingActiveTasksToCreate() {
+        return pendingActiveTasksToCreate;
+    }
+
+    Map<TaskId, Set<TopicPartition>> pendingStandbyTasksToCreate() {
+        return pendingStandbyTasksToCreate;
+    }
 }

streams/src/test/java/org/apache/kafka/streams/processor/internals/TasksTest.java

@@ -207,4 +207,27 @@ public void shouldClearFailedTask() {
         tasks.addTask(activeTask1);
         assertTrue(tasks.allNonFailedTasks().contains(activeTask1));
     }
+
+    @Test
+    public void shouldClearAllPendingTasks() {
+        final StandbyTask task = standbyTask(TASK_0_0, Set.of(TOPIC_PARTITION_B_0))
+            .inState(State.CREATED).build();
+        tasks.addPendingTasksToInit(Collections.singleton(task));
+        final TaskId taskId1 = new TaskId(0, 0, "A");
+        tasks.addPendingActiveTasksToCreate(mkMap(
+            mkEntry(taskId1, Set.of(TOPIC_PARTITION_A_0))
+        ));
+        final TaskId taskId2 = new TaskId(0, 1, "A");
+        tasks.addPendingStandbyTasksToCreate(mkMap(
+            mkEntry(taskId2, Set.of(TOPIC_PARTITION_A_0))
+        ));
+
+        assertTrue(tasks.pendingTasksToInit().contains(task));
+        assertTrue(tasks.pendingActiveTasksToCreate().containsKey(taskId1));
+        assertTrue(tasks.pendingStandbyTasksToCreate().containsKey(taskId2));
+        tasks.clear();
+        assertTrue(tasks.pendingTasksToInit().isEmpty());
+        assertTrue(tasks.pendingActiveTasksToCreate().isEmpty());
+        assertTrue(tasks.pendingStandbyTasksToCreate().isEmpty());
+    }
 }