From 987b979bdc9a791f2c28107c854d18bbcb0209ba Mon Sep 17 00:00:00 2001
From: wudi <676366545@qq.com>
Date: Tue, 9 Apr 2024 15:55:59 +0800
Subject: [PATCH] add abort log

---
 .../flink/sink/writer/DorisStreamLoad.java      |  1 +
 .../doris/flink/sink/writer/DorisWriter.java    | 17 +++++++++++++++++
 2 files changed, 18 insertions(+)

diff --git a/flink-doris-connector/src/main/java/org/apache/doris/flink/sink/writer/DorisStreamLoad.java b/flink-doris-connector/src/main/java/org/apache/doris/flink/sink/writer/DorisStreamLoad.java
index d049a2bb9..ca42f0ef3 100644
--- a/flink-doris-connector/src/main/java/org/apache/doris/flink/sink/writer/DorisStreamLoad.java
+++ b/flink-doris-connector/src/main/java/org/apache/doris/flink/sink/writer/DorisStreamLoad.java
@@ -169,6 +169,7 @@ public void abortPreCommit(String labelPrefix, long chkID) throws Exception {
                 //  Currently, it can only be aborted based on txnid, so we must
                 //  first request a streamload based on the label to get the txnid.
                 String label = labelGenerator.generateTableLabel(startChkID);
+                LOG.info("start a check label {} to load.", label);
                 HttpPutBuilder builder = new HttpPutBuilder();
                 builder.setUrl(loadUrlStr)
                         .baseAuth(user, passwd)
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 54facc704..2b4dbc02f 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
@@ -43,8 +43,10 @@
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
@@ -133,6 +135,7 @@ public void initializeLoad(Collection<DorisWriterState> state) {
     private void abortLingeringTransactions(Collection<DorisWriterState> recoveredStates)
             throws Exception {
         List<String> alreadyAborts = new ArrayList<>();
+        Set<String> alreadyAbortTables = new HashSet<>();
         // abort label in state
         for (DorisWriterState state : recoveredStates) {
             // Todo: When the sink parallelism is reduced,
@@ -152,6 +155,7 @@ private void abortLingeringTransactions(Collection<DorisWriterState> recoveredSt
             DorisStreamLoad streamLoader = getStreamLoader(key);
             streamLoader.abortPreCommit(state.getLabelPrefix(), curCheckpointId);
             alreadyAborts.add(state.getLabelPrefix());
+            alreadyAbortTables.add(key);
         }
 
         // TODO: In a multi-table scenario, if do not restore from checkpoint,
@@ -162,6 +166,19 @@ private void abortLingeringTransactions(Collection<DorisWriterState> recoveredSt
             // abort current labelPrefix
             DorisStreamLoad streamLoader = getStreamLoader(dorisOptions.getTableIdentifier());
             streamLoader.abortPreCommit(labelPrefix, curCheckpointId);
+            alreadyAbortTables.add(dorisOptions.getTableIdentifier());
+        }
+
+        LOG.info(
+                "doris stream loader size is {}, already abort table size is {}",
+                dorisStreamLoadMap.size(),
+                alreadyAbortTables.size());
+        if (dorisStreamLoadMap.size() != alreadyAbortTables.size()) {
+            Set<String> diff = new HashSet<>(dorisStreamLoadMap.keySet());
+            diff.removeAll(alreadyAbortTables);
+            LOG.warn(
+                    "there are some tables in dorisStreamLoadMap, but they are not aborted: {}",
+                    diff);
         }
     }