fix database sync

JNSimba · Oct 8, 2023 · 53fae5a · 53fae5a
1 parent 89891f5
commit 53fae5a
Showing 1 changed file with 5 additions and 2 deletions.
diff --git a/flink-doris-connector/src/main/java/org/apache/doris/flink/tools/cdc/DatabaseSync.java b/flink-doris-connector/src/main/java/org/apache/doris/flink/tools/cdc/DatabaseSync.java
@@ -190,9 +190,12 @@ public DorisSink<String> buildDorisSink(String table) {
         sinkConfig.getOptional(DorisConfigOptions.SINK_MAX_RETRIES).ifPresent(executionBuilder::setMaxRetries);
         sinkConfig.getOptional(DorisConfigOptions.SINK_IGNORE_UPDATE_BEFORE).ifPresent(executionBuilder::setIgnoreUpdateBefore);
 
-        boolean enable2pc = sinkConfig.getBoolean(DorisConfigOptions.SINK_ENABLE_2PC);
-        if(!enable2pc){
+
+        if(!sinkConfig.getBoolean(DorisConfigOptions.SINK_ENABLE_2PC)){
             executionBuilder.disable2PC();
+        } else if(sinkConfig.getOptional(DorisConfigOptions.SINK_ENABLE_2PC).isPresent()){
+            //force open 2pc
+            executionBuilder.enable2PC();
         }
 
         //batch option