fix stop load bug

JNSimba · Nov 21, 2023 · 1911b2d · 1911b2d
1 parent 65e1a14
commit 1911b2d
Showing 1 changed file with 6 additions and 2 deletions.
diff --git a/flink-doris-connector/src/main/java/org/apache/doris/flink/sink/writer/DorisWriter.java b/flink-doris-connector/src/main/java/org/apache/doris/flink/sink/writer/DorisWriter.java
@@ -209,13 +209,15 @@ public Collection<DorisCommittable> prepareCommit() throws IOException, Interrup
         }
         // disable exception checker before stop load.
         globalLoading = false;
-        // clean loadingMap
-        loadingMap.clear();
 
         // submit stream load http request
         List<DorisCommittable> committableList = new ArrayList<>();
         for(Map.Entry<String, DorisStreamLoad> streamLoader : dorisStreamLoadMap.entrySet()){
             String tableIdentifier = streamLoader.getKey();
+            if(!loadingMap.getOrDefault(tableIdentifier, false)){
+                LOG.debug("skip table {}, no data need to load.", tableIdentifier);
+                continue;
+            }
             DorisStreamLoad dorisStreamLoad = streamLoader.getValue();
             LabelGenerator labelGenerator = getLabelGenerator(tableIdentifier);
             String currentLabel = labelGenerator.generateTableLabel(curCheckpointId);
@@ -229,6 +231,8 @@ public Collection<DorisCommittable> prepareCommit() throws IOException, Interrup
                 committableList.add(new DorisCommittable(dorisStreamLoad.getHostPort(), dorisStreamLoad.getDb(), txnId));
             }
         }
+        // clean loadingMap
+        loadingMap.clear();
         return committableList;
     }