[flink] Fix performance issue in ContinuousFileSplitEnumerator (#3071)

apache · Mar 25, 2024 · d77b75d · d77b75d
1 parent 76abb84
commit d77b75d
Showing 1 changed file with 2 additions and 1 deletion.
diff --git a/...nk-common/src/main/java/org/apache/paimon/flink/source/ContinuousFileSplitEnumerator.java b/...nk-common/src/main/java/org/apache/paimon/flink/source/ContinuousFileSplitEnumerator.java
@@ -245,9 +245,10 @@ protected synchronized void assignSplits() {
         // create assignment
         Map<Integer, List<FileStoreSourceSplit>> assignment = new HashMap<>();
         Iterator<Integer> readersAwait = readersAwaitingSplit.iterator();
+        Set<Integer> subtaskIds = context.registeredReaders().keySet();
         while (readersAwait.hasNext()) {
             Integer task = readersAwait.next();
-            if (!context.registeredReaders().containsKey(task)) {
+            if (!subtaskIds.contains(task)) {
                 readersAwait.remove();
                 continue;
             }