nit

Author: xx789633

fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/maintenance/IcebergRewriteITCase.java

@@ -56,41 +56,34 @@ protected static void beforeAll() {
 
     @Test
     void testCompaction() throws Exception {
-        // create a pk table, write some records and wait until snapshot finished
-        TablePath t1 = TablePath.of(DEFAULT_DB, "pkTable");
-        long t1Id = createPkTable(t1);
-        TableBucket t1Bucket = new TableBucket(t1Id, 0);
-        // write records
-        writeFullTypeRow(t1, 1, 3);
-        waitUntilSnapshot(t1Id, 1, 0);
-
-        // then start tiering job
         JobClient jobClient = buildTieringJob(execEnv);
         try {
-            // check the status of replica after synced
-            assertReplicaStatus(t1Bucket, 3);
+            TablePath t1 = TablePath.of(DEFAULT_DB, "pkTable");
+            long t1Id = createPkTable(t1);
+            TableBucket t1Bucket = new TableBucket(t1Id, 0);
 
-            writeFullTypeRow(t1, 4, 6);
-            waitUntilSnapshot(t1Id, 1, 1);
-            assertReplicaStatus(t1Bucket, 6);
+            writeFullTypeRows(t1, t1Bucket, 1, 3, 3);
 
-            writeFullTypeRow(t1, 7, 9);
-            waitUntilSnapshot(t1Id, 1, 2);
-            assertReplicaStatus(t1Bucket, 9);
+            writeFullTypeRows(t1, t1Bucket, 4, 6, 6);
 
+            writeFullTypeRows(t1, t1Bucket, 7, 9, 9);
             checkFileCountInIcebergTable(t1, 3);
 
-            writeFullTypeRow(t1, 10, 12);
-            waitUntilSnapshot(t1Id, 1, 3);
-            assertReplicaStatus(t1Bucket, 12);
-
+            // trigger compaction
+            writeFullTypeRows(t1, t1Bucket, 10, 12, 12);
             checkFileCountInIcebergTable(t1, 2);
         } finally {
             jobClient.cancel().get();
         }
     }
 
-    private void writeFullTypeRow(TablePath tablePath, int from, int to) throws Exception {
+    private void writeFullTypeRows(
+            TablePath tablePath,
+            TableBucket tableBucket,
+            int from,
+            int to,
+            long expectedLogEndOffset)
+            throws Exception {
         List<InternalRow> rows = new ArrayList<>();
         for (int i = from; i <= to; i++) {
             rows.add(
@@ -116,5 +109,6 @@ private void writeFullTypeRow(TablePath tablePath, int from, int to) throws Exce
                             new byte[] {1, 2, 3, 4, 5, 6, 7, 8, 9, 10}));
         }
         writeRows(tablePath, rows, false);
+        assertReplicaStatus(tableBucket, expectedLogEndOffset);
     }
 }

fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/testutils/FlinkIcebergTieringTestBase.java

@@ -387,7 +387,7 @@ protected void checkDataInIcebergAppendOnlyTable(
         }
     }
 
-    protected void checkFileInIcebergTable(TablePath tablePath, int expectedFileCount)
+    protected void checkFileCountInIcebergTable(TablePath tablePath, int expectedFileCount)
             throws IOException {
         org.apache.iceberg.Table table = icebergCatalog.loadTable(toIceberg(tablePath));
         int count = 0;