[flink] Respect waitCompaction in FlinkCdcMultiTableSink (apache#3337)

tongcheng-elong · May 15, 2024 · 3b006fe · 3b006fe
1 parent b16af31
commit 3b006fe
Showing 1 changed file with 3 additions and 1 deletion.
diff --git a/...imon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/FlinkCdcMultiTableSink.java b/...imon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/FlinkCdcMultiTableSink.java
@@ -19,6 +19,7 @@
 package org.apache.paimon.flink.sink.cdc;
 
 import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.flink.FlinkConnectorOptions;
 import org.apache.paimon.flink.VersionedSerializerWrapper;
 import org.apache.paimon.flink.sink.CommittableStateManager;
 import org.apache.paimon.flink.sink.Committer;
@@ -86,7 +87,8 @@ private StoreSinkWrite.WithWriteBufferProvider createWriteProvider() {
                         state,
                         ioManager,
                         isOverwrite,
-                        false,
+                        FlinkConnectorOptions.prepareCommitWaitCompaction(
+                                table.coreOptions().toConfiguration()),
                         true,
                         memoryPoolFactory,
                         metricGroup);