[Flink] Fix potential dead loop in source enumerator (lakesoul-io#560)

* fix potential dead loop in source enumerator Signed-off-by: chenxu <[email protected]> * fix spark/flink deps Signed-off-by: chenxu <[email protected]> * add synchronized block Signed-off-by: chenxu <[email protected]> * fix commons-io conflict Signed-off-by: chenxu <[email protected]> * fix commons-csv version Signed-off-by: chenxu <[email protected]> --------- Signed-off-by: chenxu <[email protected]> Co-authored-by: chenxu <[email protected]>
xuchen-plus · Nov 20, 2024 · 366beb8 · 366beb8
1 parent 60a55e9
commit 366beb8
Show file tree

Hide file tree

Showing 5 changed files with 65 additions and 20 deletions.
diff --git a/lakesoul-common/pom.xml b/lakesoul-common/pom.xml
@@ -309,8 +309,17 @@ SPDX-License-Identifier: Apache-2.0
                     <groupId>org.slf4j</groupId>
                     <artifactId>slf4j-api</artifactId>
                 </exclusion>
+                <exclusion>
+                    <groupId>org.casbin</groupId>
+                    <artifactId>jcasbin</artifactId>
+                </exclusion>
             </exclusions>
         </dependency>
+        <dependency>
+            <groupId>org.casbin</groupId>
+            <artifactId>jcasbin</artifactId>
+            <version>1.73.0</version>
+        </dependency>
         <!-- aspectJ -->
         <dependency>
             <groupId>org.aspectj</groupId>

diff --git a/lakesoul-flink/pom.xml b/lakesoul-flink/pom.xml
@@ -41,6 +41,18 @@ SPDX-License-Identifier: Apache-2.0
                     <groupId>org.slf4j</groupId>
                     <artifactId>slf4j-api</artifactId>
                 </exclusion>
+                <exclusion>
+                    <groupId>com.fasterxml.jackson.core</groupId>
+                    <artifactId>*</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>commons-io</groupId>
+                    <artifactId>commons-io</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>commons-codec</groupId>
+                    <artifactId>commons-codec</artifactId>
+                </exclusion>
             </exclusions>
         </dependency>
         <dependency>

diff --git a/...ain/java/org/apache/flink/lakesoul/source/LakeSoulAllPartitionDynamicSplitEnumerator.java b/...ain/java/org/apache/flink/lakesoul/source/LakeSoulAllPartitionDynamicSplitEnumerator.java
@@ -90,20 +90,23 @@ public void handleSplitRequest(int subtaskId, @Nullable String requesterHostname
             return;
         }
         int tasksSize = context.registeredReaders().size();
-        Optional<LakeSoulPartitionSplit> nextSplit = this.splitAssigner.getNext(subtaskId, tasksSize);
-        if (nextSplit.isPresent()) {
-            context.assignSplit(nextSplit.get(), subtaskId);
-            taskIdsAwaitingSplit.remove(subtaskId);
-        } else {
-            taskIdsAwaitingSplit.add(subtaskId);
+        synchronized (this) {
+            Optional<LakeSoulPartitionSplit> nextSplit = this.splitAssigner.getNext(subtaskId, tasksSize);
+            if (nextSplit.isPresent()) {
+                context.assignSplit(nextSplit.get(), subtaskId);
+                taskIdsAwaitingSplit.remove(subtaskId);
+            } else {
+                taskIdsAwaitingSplit.add(subtaskId);
+            }
         }
-
     }
 
     @Override
     public void addSplitsBack(List<LakeSoulPartitionSplit> splits, int subtaskId) {
         LOG.info("Add split back: {}", splits);
-        splitAssigner.addSplits(splits);
+        synchronized (this) {
+            splitAssigner.addSplits(splits);
+        }
     }
 
     @Override
@@ -112,11 +115,13 @@ public void addReader(int subtaskId) {
 
     @Override
     public LakeSoulPendingSplits snapshotState(long checkpointId) throws Exception {
-        LakeSoulPendingSplits pendingSplits =
-                new LakeSoulPendingSplits(splitAssigner.remainingSplits(), this.nextStartTime, this.tableId, "",
-                        this.discoveryInterval, this.hashBucketNum);
-        LOG.info("LakeSoulAllPartitionDynamicSplitEnumerator snapshotState {}", pendingSplits);
-        return pendingSplits;
+        synchronized (this) {
+            LakeSoulPendingSplits pendingSplits =
+                    new LakeSoulPendingSplits(splitAssigner.remainingSplits(), this.nextStartTime, this.tableId, "",
+                            this.discoveryInterval, this.hashBucketNum);
+            LOG.info("LakeSoulAllPartitionDynamicSplitEnumerator snapshotState {}", pendingSplits);
+            return pendingSplits;
+        }
     }
 
     @Override
@@ -131,12 +136,16 @@ private void processDiscoveredSplits(Collection<LakeSoulPartitionSplit> splits,
         }
         LOG.info("Process discovered splits {}", splits);
         int tasksSize = context.registeredReaders().size();
-        this.splitAssigner.addSplits(splits);
-        for (Integer item : taskIdsAwaitingSplit) {
-            Optional<LakeSoulPartitionSplit> al = this.splitAssigner.getNext(item, tasksSize);
-            if (al.isPresent()) {
-                context.assignSplit(al.get(), item);
-                taskIdsAwaitingSplit.remove(item);
+        synchronized (this) {
+            this.splitAssigner.addSplits(splits);
+            Iterator<Integer> iter = taskIdsAwaitingSplit.iterator();
+            while (iter.hasNext()) {
+                int taskId = iter.next();
+                Optional<LakeSoulPartitionSplit> al = this.splitAssigner.getNext(taskId, tasksSize);
+                if (al.isPresent()) {
+                    context.assignSplit(al.get(), taskId);
+                    iter.remove();
+                }
             }
         }
     }
@@ -147,7 +156,6 @@ public Collection<LakeSoulPartitionSplit> enumerateSplits() {
         List<PartitionInfo> filteredPartition = SubstraitUtil.applyPartitionFilters(allPartitionInfo, partitionArrowSchema, partitionFilters);
         LOG.info("filteredPartition={}, filter={}", filteredPartition, partitionFilters);
 
-
         ArrayList<LakeSoulPartitionSplit> splits = new ArrayList<>(16);
         for (PartitionInfo partitionInfo : filteredPartition) {
             String partitionDesc = partitionInfo.getPartitionDesc();

diff --git a/lakesoul-spark/pom.xml b/lakesoul-spark/pom.xml
@@ -63,6 +63,14 @@ SPDX-License-Identifier: Apache-2.0
                     <groupId>com.fasterxml.jackson.core</groupId>
                     <artifactId>*</artifactId>
                 </exclusion>
+                <exclusion>
+                    <groupId>commons-io</groupId>
+                    <artifactId>commons-io</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>commons-codec</groupId>
+                    <artifactId>commons-codec</artifactId>
+                </exclusion>
             </exclusions>
         </dependency>
         <dependency>

diff --git a/native-io/lakesoul-io-java/pom.xml b/native-io/lakesoul-io-java/pom.xml
@@ -144,6 +144,14 @@ SPDX-License-Identifier: Apache-2.0
                     <groupId>com.fasterxml.jackson.core</groupId>
                     <artifactId>*</artifactId>
                 </exclusion>
+                <exclusion>
+                    <groupId>com.fasterxml.jackson.core</groupId>
+                    <artifactId>jackson-annotations</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>com.fasterxml.jackson.core</groupId>
+                    <artifactId>jackson-core</artifactId>
+                </exclusion>
             </exclusions>
             <version>${substrait.version}</version>
             <scope>compile</scope>