improve some log

JNSimba · Jan 3, 2025 · 0c4a50f · 0c4a50f
1 parent 1023235
commit 0c4a50f
Show file tree

Hide file tree

Showing 3 changed files with 12 additions and 6 deletions.
diff --git a/...k-doris-connector/src/main/java/org/apache/doris/flink/sink/committer/DorisCommitter.java b/...k-doris-connector/src/main/java/org/apache/doris/flink/sink/committer/DorisCommitter.java
@@ -128,12 +128,16 @@ private void commitTransaction(DorisCommittable committable) throws IOException
                                 jsonMapper.readValue(
                                         loadResult,
                                         new TypeReference<HashMap<String, String>>() {});
-                        if (!res.get("status").equals(SUCCESS)
-                                && !ResponseUtil.isCommitted(res.get("msg"))) {
+                        if (res.get("status").equals(SUCCESS)) {
+                            LOG.info("load result {}", loadResult);
+                        } else if (ResponseUtil.isCommitted(res.get("msg"))) {
+                            LOG.info(
+                                    "transaction {} has already committed successfully, skipping, load response is {}",
+                                    committable.getTxnID(),
+                                    res.get("msg"));
+                        } else {
                             throw new DorisRuntimeException(
                                     "commit transaction failed " + loadResult);
-                        } else {
-                            LOG.info("load result {}", loadResult);
                         }
                         return;
                     }

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
@@ -391,7 +391,10 @@ public void abortTransaction(long txnID) throws Exception {
             String msg = res.get("msg");
             // transaction already aborted
             if (msg != null && ResponseUtil.isAborted(msg)) {
-                LOG.warn("Failed to abort transaction, {}", msg);
+                LOG.info(
+                        "transaction {} may have already been successfully aborted, skipping, abort response is {}",
+                        txnID,
+                        msg);
                 return;
             }
 

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
@@ -458,7 +458,6 @@ public void close() throws Exception {
         if (scheduledExecutorService != null) {
             scheduledExecutorService.shutdownNow();
         }
-        LOG.info("Try to abort txn before closing.");
         abortPossibleSuccessfulTransaction();
 
         if (dorisStreamLoadMap != null && !dorisStreamLoadMap.isEmpty()) {