(paimon) branch master updated: [flink] Fix unstable changelog compaction test (#9281)

[lzljs3620320]

This is an automated email from the ASF dual-hosted git repository.

JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git


The following commit(s) were added to refs/heads/master by this push:
     new 4af1470c52 [flink] Fix unstable changelog compaction test (#9281)
4af1470c52 is described below

commit 4af1470c521960789f81b381cc15a9483282ff29
Author: QuakeWang <45645138+quakew...@users.noreply.github.com>
AuthorDate: Tue Aug 18 11:47:26 2026 +0800

    [flink] Fix unstable changelog compaction test (#9281)
---
 .../paimon/flink/PrimaryKeyFileStoreTableITCase.java  |  4 ++--
 .../compact/changelog/ChangelogCompactTaskTest.java   | 19 +++++++++++++++++++
 2 files changed, 21 insertions(+), 2 deletions(-)

diff --git 
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PrimaryKeyFileStoreTableITCase.java
 
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PrimaryKeyFileStoreTableITCase.java
index 87d1113816..20fcc94c27 100644
--- 
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PrimaryKeyFileStoreTableITCase.java
+++ 
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PrimaryKeyFileStoreTableITCase.java
@@ -1187,7 +1187,8 @@ public class PrimaryKeyFileStoreTableITCase extends 
AbstractTestBase {
 
         List<String> compactedChangelogs2 = 
listAllFilesWithPrefix("compacted-changelog-");
         assertThat(compactedChangelogs2).hasSize(2);
-        assertThat(listAllFilesWithPrefix("changelog-")).isEmpty();
+        // A single changelog file in a partition is intentionally passed 
through, so original
+        // changelog files may remain when input crosses checkpoint boundaries.
 
         // write update data
         values.clear();
@@ -1207,7 +1208,6 @@ public class PrimaryKeyFileStoreTableITCase extends 
AbstractTestBase {
         }
         assertStreamingResult(it, expected.subList(200, 600));
         assertThat(listAllFilesWithPrefix("compacted-changelog-")).hasSize(4);
-        assertThat(listAllFilesWithPrefix("changelog-")).isEmpty();
     }
 
     private List<String> listAllFilesWithPrefix(String prefix) throws 
Exception {
diff --git 
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/compact/changelog/ChangelogCompactTaskTest.java
 
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/compact/changelog/ChangelogCompactTaskTest.java
index a7e7fc099b..715fe52dda 100644
--- 
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/compact/changelog/ChangelogCompactTaskTest.java
+++ 
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/compact/changelog/ChangelogCompactTaskTest.java
@@ -27,6 +27,7 @@ import org.apache.paimon.data.InternalRow;
 import org.apache.paimon.fs.Path;
 import org.apache.paimon.fs.local.LocalFileIO;
 import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataFilePathFactory;
 import org.apache.paimon.options.MemorySize;
 import org.apache.paimon.reader.RecordReaderIterator;
 import org.apache.paimon.schema.Schema;
@@ -163,6 +164,20 @@ public class ChangelogCompactTaskTest {
         }
         write.close();
 
+        List<Path> inputFiles = new ArrayList<>();
+        for (Map.Entry<Integer, List<DataFileMeta>> entry : files.entrySet()) {
+            DataFilePathFactory pathFactory =
+                    table.store()
+                            .pathFactory()
+                            .createDataFilePathFactory(BinaryRow.EMPTY_ROW, 
entry.getKey());
+            for (DataFileMeta file : entry.getValue()) {
+                inputFiles.add(pathFactory.toPath(file));
+            }
+        }
+        for (Path file : inputFiles) {
+            assertThat(table.fileIO().exists(file)).isTrue();
+        }
+
         ChangelogCompactTask task =
                 new ChangelogCompactTask(1, BinaryRow.EMPTY_ROW, 2, files, new 
HashMap<>());
 
@@ -171,6 +186,10 @@ public class ChangelogCompactTaskTest {
                         .stream()
                         .map(c -> (CommitMessageImpl) c.commitMessage())
                         .collect(Collectors.toList());
+        for (Path file : inputFiles) {
+            assertThat(table.fileIO().exists(file)).isFalse();
+        }
+
         TableCommitImpl commit = table.newCommit("test");
         commit.commit(messages);
         commit.close();