From 9e682392d79aa607b68856cc93f761aedeb32dd4 Mon Sep 17 00:00:00 2001
From: wudi <676366545@qq.com>
Date: Fri, 3 Nov 2023 10:15:25 +0800
Subject: [PATCH] [improve] commit message compatible (#220)

---
 .../java/org/apache/doris/flink/sink/ResponseUtil.java    | 8 +++-----
 .../org/apache/doris/flink/sink/writer/DorisWriter.java   | 3 +--
 .../doris/flink/table/DorisDynamicTableFactory.java       | 1 +
 3 files changed, 5 insertions(+), 7 deletions(-)

diff --git a/flink-doris-connector/src/main/java/org/apache/doris/flink/sink/ResponseUtil.java b/flink-doris-connector/src/main/java/org/apache/doris/flink/sink/ResponseUtil.java
index 1e9a9c059..8839a1a23 100644
--- a/flink-doris-connector/src/main/java/org/apache/doris/flink/sink/ResponseUtil.java
+++ b/flink-doris-connector/src/main/java/org/apache/doris/flink/sink/ResponseUtil.java
@@ -24,13 +24,11 @@
  */
 public class ResponseUtil {
     public static final Pattern LABEL_EXIST_PATTERN =
-            Pattern.compile("errCode = 2, detailMessage = Label \\[(.*)\\] " +
-                    "has already been used, relate to txn \\[(\\d+)\\]");
+            Pattern.compile("Label \\[(.*)\\] has already been used, relate to txn \\[(\\d+)\\]");
     public static final Pattern COMMITTED_PATTERN =
-            Pattern.compile("errCode = 2, detailMessage = transaction \\[(\\d+)\\] " +
-                    "is already \\b(COMMITTED|committed|VISIBLE|visible)\\b, not pre-committed.");
+            Pattern.compile("transaction \\[(\\d+)\\] is already \\b(COMMITTED|committed|VISIBLE|visible)\\b, not pre-committed.");
 
     public static boolean isCommitted(String msg) {
-       return COMMITTED_PATTERN.matcher(msg).matches();
+       return COMMITTED_PATTERN.matcher(msg).find();
     }
 }
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 295a0be86..b61d174bf 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
@@ -33,7 +33,6 @@
 import org.apache.flink.api.connector.sink2.StatefulSink;
 import org.apache.flink.api.connector.sink2.TwoPhaseCommittingSink;
 import org.apache.flink.runtime.checkpoint.CheckpointIDCounter;
-import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableList;
 import org.apache.flink.util.Preconditions;
 import org.apache.flink.util.concurrent.ExecutorThreadFactory;
 import org.slf4j.Logger;
@@ -171,7 +170,7 @@ public Collection<DorisCommittable> prepareCommit() throws IOException, Interrup
             return Collections.emptyList();
         }
         long txnId = respContent.getTxnId();
-        return ImmutableList.of(new DorisCommittable(dorisStreamLoad.getHostPort(), dorisStreamLoad.getDb(), txnId));
+        return Collections.singletonList(new DorisCommittable(dorisStreamLoad.getHostPort(), dorisStreamLoad.getDb(), txnId));
     }
 
     @Override
diff --git a/flink-doris-connector/src/main/java/org/apache/doris/flink/table/DorisDynamicTableFactory.java b/flink-doris-connector/src/main/java/org/apache/doris/flink/table/DorisDynamicTableFactory.java
index 978fa2703..bccb8b7f3 100644
--- a/flink-doris-connector/src/main/java/org/apache/doris/flink/table/DorisDynamicTableFactory.java
+++ b/flink-doris-connector/src/main/java/org/apache/doris/flink/table/DorisDynamicTableFactory.java
@@ -111,6 +111,7 @@ public Set<ConfigOption<?>> optionalOptions() {
         options.add(USERNAME);
         options.add(PASSWORD);
         options.add(JDBC_URL);
+        options.add(AUTO_REDIRECT);
 
         options.add(DORIS_READ_FIELD);
         options.add(DORIS_FILTER_QUERY);