update

JNSimba · Sep 10, 2024 · d3143f8 · d3143f8
1 parent 5f7ad3f
commit d3143f8
Show file tree

Hide file tree

Showing 2 changed files with 11 additions and 13 deletions.
diff --git a/...-connector/src/main/java/org/apache/doris/flink/source/assigners/SimpleSplitAssigner.java b/...-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
@@ -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) {