KAFKA-12185: fix ConcurrentModificationException in newly added Tasks…

… container class (apache#9940) Reviewers: Guozhang Wang <[email protected]>, A. Sophie Blee-Goldman <[email protected]>
mghildiy · Jan 21, 2021 · 92e72f7 · 92e72f7
1 parent 24e7e81
commit 92e72f7
Show file tree

Hide file tree

Showing 2 changed files with 115 additions and 59 deletions.
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/Tasks.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/Tasks.java
@@ -128,9 +128,11 @@ void convertActiveToStandby(final StreamTask activeTask,
         if (activeTasksPerId.remove(activeTask.id()) == null) {
             throw new IllegalStateException("Attempted to convert unknown active task to standby task: " + activeTask.id());
         }
-        activeTasksPerPartition.entrySet().stream()
+        final Set<TopicPartition> toBeRemoved = activeTasksPerPartition.entrySet().stream()
             .filter(e -> e.getValue().id().equals(activeTask.id()))
-            .forEach(e -> activeTasksPerPartition.remove(e.getKey()));
+            .map(Map.Entry::getKey)
+            .collect(Collectors.toSet());
+        toBeRemoved.forEach(activeTasksPerPartition::remove);
 
         cleanUpTaskProducerAndRemoveTask(activeTask.id(), taskCloseExceptions);