root
/
kafka
mirror of https://github.com/apache/kafka.git
1
0
Fork 0
Code Issues Actions 2 Packages Projects Releases Wiki Activity

MINOR: clear pendingTasksToInit on tasks clear.

This commit is contained in:
Nikita 2025-10-06 21:57:32 -07:00
parent f5a87b3703
commit bf31d613f5
No known key found for this signature in database
2 changed files with 12 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -258,6 +258,7 @@ class Tasks implements TasksRegistry {
@Override @Override
public synchronized void clear() { public synchronized void clear() {
pendingTasksToInit.clear();
activeTasksPerId.clear(); activeTasksPerId.clear();
standbyTasksPerId.clear(); standbyTasksPerId.clear();
activeTasksPerPartition.clear(); activeTasksPerPartition.clear();

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

@ -207,4 +207,15 @@ public class TasksTest {
tasks.addTask(activeTask1); tasks.addTask(activeTask1);
assertTrue(tasks.allNonFailedTasks().contains(activeTask1)); assertTrue(tasks.allNonFailedTasks().contains(activeTask1));
} }
@Test
public void shouldClearPendingToInitTasks() {
final StandbyTask task = standbyTask(TASK_0_0, Set.of(TOPIC_PARTITION_B_0))
.inState(State.CREATED).build();
tasks.addPendingTasksToInit(Collections.singleton(task));
assertTrue(tasks.pendingTasksToInit().contains(task));
tasks.clear();
assertFalse(tasks.pendingTasksToInit().contains(task));
}
} }