[fix] Fix error in FlinkSink for changelog files merge (#4415)

This closes #4415.
apache · Oct 31, 2024 · 092ebe0 · 092ebe0
1 parent 734a0dc
commit 092ebe0
Show file tree

Hide file tree

Showing 2 changed files with 2 additions and 2 deletions.
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSink.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSink.java
@@ -232,7 +232,7 @@ public DataStream<Committable> doWrite(
             declareManagedMemory(written, options.get(SINK_MANAGED_WRITER_BUFFER_MEMORY));
         }
 
-        if (options.contains(CHANGELOG_PRECOMMIT_COMPACT)) {
+        if (options.get(CHANGELOG_PRECOMMIT_COMPACT)) {
             written =
                     written.transform(
                                     "Changelog Compact Coordinator",

diff --git a/...on-flink-common/src/test/java/org/apache/paimon/flink/PrimaryKeyFileStoreTableITCase.java b/...on-flink-common/src/test/java/org/apache/paimon/flink/PrimaryKeyFileStoreTableITCase.java
@@ -688,7 +688,7 @@ private void testLookupChangelogProducerRandom(
                                 + "'lookup-wait' = '%s', "
                                 + "'deletion-vectors.enabled' = '%s', "
                                 + "'changelog.precommit-compact' = '%s'",
-                        random.nextBoolean() ? "512kb" : "1mb",
+                        random.nextBoolean() ? "4mb" : "8mb",
                         random.nextBoolean(),
                         enableDeletionVectors,
                         random.nextBoolean()));