[hotfix] Fix test PaimonTieringTest.testTieringWriteTable is unstable. (#1034)

Author: zhuangchong

fluss-lake/fluss-lake-paimon/src/test/java/com/alibaba/fluss/lake/paimon/tiering/PaimonTieringTest.java

@@ -266,20 +266,35 @@ private Tuple2<List<LogRecord>, List<LogRecord>> genPrimaryKeyTableRecords(
                         toRecord(++offset, rows.get(0), INSERT),
                         toRecord(++offset, rows.get(1), UPDATE_BEFORE),
                         toRecord(++offset, rows.get(2), UPDATE_AFTER)));
-        expectLogRecords.add(toRecord(offset, rows.get(2), UPDATE_AFTER));
+        expectLogRecords.add(
+                toRecord(
+                        offset,
+                        writtenLogRecords.get(writtenLogRecords.size() - 1).timestamp(),
+                        rows.get(2),
+                        UPDATE_AFTER));
 
         // gen +I, +U
         rows = genKvRow(partition, bucket, 2, 7, 9);
         writtenLogRecords.addAll(
                 Arrays.asList(
                         toRecord(++offset, rows.get(0), INSERT),
                         toRecord(++offset, rows.get(1), UPDATE_AFTER)));
-        expectLogRecords.add(toRecord(offset, rows.get(1), UPDATE_AFTER));
+        expectLogRecords.add(
+                toRecord(
+                        offset,
+                        writtenLogRecords.get(writtenLogRecords.size() - 1).timestamp(),
+                        rows.get(1),
+                        UPDATE_AFTER));
 
         // gen +I
         rows = genKvRow(partition, bucket, 3, 9, 10);
         writtenLogRecords.add(toRecord(++offset, rows.get(0), INSERT));
-        expectLogRecords.add(toRecord(offset, rows.get(0), INSERT));
+        expectLogRecords.add(
+                toRecord(
+                        offset,
+                        writtenLogRecords.get(writtenLogRecords.size() - 1).timestamp(),
+                        rows.get(0),
+                        INSERT));
 
         return Tuple2.of(writtenLogRecords, expectLogRecords);
     }
@@ -302,7 +317,12 @@ private List<GenericRow> genKvRow(
     }
 
     private GenericRecord toRecord(long offset, GenericRow row, ChangeType changeType) {
-        return new GenericRecord(offset, System.currentTimeMillis(), changeType, row);
+        return toRecord(offset, System.currentTimeMillis(), row, changeType);
+    }
+
+    private GenericRecord toRecord(
+            long offset, long timestamp, GenericRow row, ChangeType changeType) {
+        return new GenericRecord(offset, timestamp, changeType, row);
     }
 
     private CloseableIterator<InternalRow> getPaimonRows(