From d3143f87aa8d78fef378eb0e92ba95a7da9b5407 Mon Sep 17 00:00:00 2001
From: wudi <676366545@qq.com>
Date: Tue, 10 Sep 2024 16:04:50 +0800
Subject: [PATCH] update

---
 .../source/assigners/SimpleSplitAssigner.java |  2 +-
 .../doris/flink/source/DorisSourceITCase.java | 22 +++++++++----------
 2 files changed, 11 insertions(+), 13 deletions(-)

diff --git a/flink-doris-connector/src/main/java/org/apache/doris/flink/source/assigners/SimpleSplitAssigner.java b/flink-doris-connector/src/main/java/org/apache/doris/flink/source/assigners/SimpleSplitAssigner.java
index 52afa89b0..3409f6038 100644
--- a/flink-doris-connector/src/main/java/org/apache/doris/flink/source/assigners/SimpleSplitAssigner.java
+++ b/flink-doris-connector/src/main/java/org/apache/doris/flink/source/assigners/SimpleSplitAssigner.java
@@ -52,7 +52,7 @@ public void addSplits(Collection<DorisSourceSplit> splits) {
 
     @Override
     public PendingSplitsCheckpoint snapshotState(long checkpointId) {
-        LOG.info("Checkpointing {} splits: {}", checkpointId, splits);
+        LOG.info("Checkpointing splits: {}, id {}", splits, checkpointId);
         return new PendingSplitsCheckpoint(splits);
     }
 
diff --git a/flink-doris-connector/src/test/java/org/apache/doris/flink/source/DorisSourceITCase.java b/flink-doris-connector/src/test/java/org/apache/doris/flink/source/DorisSourceITCase.java
index aaf973e43..5a9c200f8 100644
--- a/flink-doris-connector/src/test/java/org/apache/doris/flink/source/DorisSourceITCase.java
+++ b/flink-doris-connector/src/test/java/org/apache/doris/flink/source/DorisSourceITCase.java
@@ -348,7 +348,7 @@ public void testJobManagerFailoverSource() throws Exception {
         initializeTableWithData(TABLE_CSV_JM);
         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
         env.setParallelism(DEFAULT_PARALLELISM);
-        env.enableCheckpointing(100L);
+        env.enableCheckpointing(200L);
         env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 0));
         StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
         String sourceDDL =
@@ -372,7 +372,7 @@ public void testJobManagerFailoverSource() throws Exception {
         CloseableIterator<Row> iterator = tableResult.collect();
         JobID jobId = tableResult.getJobClient().get().getJobID();
 
-        List<String> expectedSnapshotData = getExpectedData();
+        List<String> expectedData = getExpectedData();
         if (iterator.hasNext()) {
             LOG.info("trigger jobmanager failover...");
             triggerFailover(
@@ -381,10 +381,9 @@ public void testJobManagerFailoverSource() throws Exception {
                     miniClusterResource.getMiniCluster(),
                     () -> sleepMs(100));
         }
-        checkResultInAnyOrder(
-                "testJobManagerFailoverSource",
-                expectedSnapshotData.toArray(),
-                fetchRows(iterator).toArray());
+        List<String> actual = fetchRows(iterator);
+        LOG.info("actual data: {}, expected: {}", actual, expectedData);
+        Assert.assertTrue(actual.containsAll(expectedData));
     }
 
     private static List<String> getExpectedData() {
@@ -427,7 +426,7 @@ public void testTaskManagerFailoverSource() throws Exception {
         initializeTableWithData(TABLE_CSV_TM);
         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
         env.setParallelism(DEFAULT_PARALLELISM);
-        env.enableCheckpointing(100L);
+        env.enableCheckpointing(200L);
         env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 0));
         StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
         String sourceDDL =
@@ -450,7 +449,7 @@ public void testTaskManagerFailoverSource() throws Exception {
         TableResult tableResult = tEnv.executeSql("select * from doris_source_tm");
         CloseableIterator<Row> iterator = tableResult.collect();
         JobID jobId = tableResult.getJobClient().get().getJobID();
-        List<String> expectedSnapshotData = getExpectedData();
+        List<String> expectedData = getExpectedData();
         if (iterator.hasNext()) {
             LOG.info("trigger taskmanager failover...");
             triggerFailover(
@@ -460,10 +459,9 @@ public void testTaskManagerFailoverSource() throws Exception {
                     () -> sleepMs(100));
         }
 
-        checkResultInAnyOrder(
-                "testTaskManagerFailoverSource",
-                expectedSnapshotData.toArray(),
-                fetchRows(iterator).toArray());
+        List<String> actual = fetchRows(iterator);
+        LOG.info("actual data: {}, expected: {}", actual, expectedData);
+        Assert.assertTrue(actual.containsAll(expectedData));
     }
 
     private void checkResult(String testName, Object[] expected, Object[] actual) {