[Bug] fix stopload bug when restore from checkpoint (#232)
If you restore from checkpoint, the checkpoint table is tb1, but the newly imported table does not have tb1. You can abort normally, but an error will be reported during stopload.
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
index 8550a21..c5ce847 100644
--- 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 @@
}
// 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 @@
committableList.add(new DorisCommittable(dorisStreamLoad.getHostPort(), dorisStreamLoad.getDb(), txnId));
}
}
+ // clean loadingMap
+ loadingMap.clear();
return committableList;
}
