[FLINK-39363][Test] Fix flaky test PaimonSinkITCase

Author: Hisoka-X

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/test/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonSinkITCase.java

@@ -88,6 +88,7 @@
 import org.apache.paimon.flink.sink.MultiTableCommittable;
 import org.apache.paimon.options.Options;
 import org.assertj.core.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.io.TempDir;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.CsvSource;
@@ -133,7 +134,12 @@ public class PaimonSinkITCase {
     private final TableId table1 = TableId.tableId("test", "table1");
     private final TableId table2 = TableId.tableId("test", "table2");
 
-    private static int checkpointId = 1;
+    private int checkpointId = 1;
+
+    @BeforeEach
+    public void resetCheckpointId() {
+        checkpointId = 1;
+    }
 
     public static final String TEST_DATABASE = "test";
     private static final String HADOOP_CONF_DIR =
@@ -756,18 +762,17 @@ public void testSinkWithMultiTables(String metastore, boolean enableDeleteVector
                 .containsExactlyInAnyOrder(Row.ofKind(RowKind.INSERT, "1", "1"));
     }
 
-    private static void commit(
-            PaimonWriter<Event> writer, Committer<MultiTableCommittable> committer)
+    private void commit(PaimonWriter<Event> writer, Committer<MultiTableCommittable> committer)
             throws IOException, InterruptedException {
         Collection<Committer.CommitRequest<MultiTableCommittable>> commitRequests =
                 writer.prepareCommit().stream()
-                        .map(PaimonSinkITCase::correctCheckpointId)
+                        .map(this::correctCheckpointId)
                         .map(MockCommitRequestImpl::new)
                         .collect(Collectors.toList());
         committer.commit(commitRequests);
     }
 
-    private static void writeAndCommit(
+    private void writeAndCommit(
             PaimonWriter<Event> writer, Committer<MultiTableCommittable> committer, Event... events)
             throws IOException, InterruptedException {
         for (Event event : events) {
@@ -777,7 +782,7 @@ private static void writeAndCommit(
         commit(writer, committer);
     }
 
-    private static void writeAndCommit(
+    private void writeAndCommit(
             BucketAssignOperator bucketAssignOperator,
             PaimonWriter<Event> writer,
             Committer<MultiTableCommittable> committer,
@@ -868,7 +873,7 @@ public void testDuplicateCommitAfterRestore(String metastore, boolean enableDele
         writer.flush(false);
         Collection<Committer.CommitRequest<MultiTableCommittable>> commitRequests =
                 writer.prepareCommit().stream()
-                        .map(PaimonSinkITCase::correctCheckpointId)
+                        .map(this::correctCheckpointId)
                         .map(MockCommitRequestImpl::new)
                         .collect(Collectors.toList());
         committer.commit(commitRequests);
@@ -900,7 +905,7 @@ public void testDuplicateCommitAfterRestore(String metastore, boolean enableDele
             // Checkpoint id start from 1
             committer.commit(
                     writer.prepareCommit().stream()
-                            .map(PaimonSinkITCase::correctCheckpointId)
+                            .map(this::correctCheckpointId)
                             .map(MockCommitRequestImpl::new)
                             .collect(Collectors.toList()));
         }
@@ -1039,7 +1044,7 @@ private void runJobWithEvents(List<Event> events, boolean isBatchMode) throws Ex
         env.execute("runJobWithEvents").getJobExecutionResult();
     }
 
-    private static MultiTableCommittable correctCheckpointId(MultiTableCommittable committable) {
+    private MultiTableCommittable correctCheckpointId(MultiTableCommittable committable) {
         // update the right checkpointId for MultiTableCommittable
         return new MultiTableCommittable(
                 committable.getDatabase(),