[improve] commit message compatible (apache#220)

JNSimba · Nov 3, 2023 · 9e68239 · 9e68239
1 parent 115017b
commit 9e68239
Show file tree

Hide file tree

Showing 3 changed files with 5 additions and 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
@@ -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
@@ -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/...-doris-connector/src/main/java/org/apache/doris/flink/table/DorisDynamicTableFactory.java b/...-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);