[e2e][tests] Improve the stability of pipeline e2e tests

apache · Aug 27, 2024 · d1d8a24 · d1d8a24
1 parent a876af2
commit d1d8a24
Show file tree

Hide file tree

Showing 3 changed files with 8 additions and 7 deletions.
diff --git a/...src/main/java/org/apache/flink/cdc/connectors/mysql/source/utils/TableDiscoveryUtils.java b/...src/main/java/org/apache/flink/cdc/connectors/mysql/source/utils/TableDiscoveryUtils.java
@@ -88,7 +88,8 @@ public static List<TableId> listTables(
                                             "\t including table '{}' for further processing",
                                             tableId);
                                 } else {
-                                    LOG.info("\t '{}' is filtered out of table capturing", tableId);
+                                    LOG.debug(
+                                            "\t '{}' is filtered out of table capturing", tableId);
                                 }
                             }
                         });

diff --git a/...-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/RouteE2eITCase.java b/...-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/RouteE2eITCase.java
@@ -24,7 +24,6 @@
 import org.apache.flink.cdc.pipeline.tests.utils.PipelineTestEnvironment;
 
 import org.junit.After;
-import org.junit.Assert;
 import org.junit.Before;
 import org.junit.ClassRule;
 import org.junit.Test;
@@ -719,6 +718,11 @@ public void testMergeTableRouteWithTransform() throws Exception {
 
     @Test
     public void testReplacementSymbol() throws Exception {
+        String defaultFlinkProperties = getFlinkProperties(flinkVersion);
+        overrideFlinkProperties(
+                defaultFlinkProperties.replace(
+                        "execution.checkpointing.interval: 300",
+                        "execution.checkpointing.interval: 10000"));
         String pipelineJob =
                 String.format(
                         "source:\n"
@@ -839,8 +843,4 @@ private void waitUntilSpecificEvent(String event) throws Exception {
                             + taskManagerConsumer.toUtf8String());
         }
     }
-
-    private void assertNotExists(String event) {
-        Assert.assertFalse(taskManagerConsumer.toUtf8String().contains(event));
-    }
 }
diff --git a/...ests/src/test/java/org/apache/flink/cdc/pipeline/tests/utils/PipelineTestEnvironment.java b/...ests/src/test/java/org/apache/flink/cdc/pipeline/tests/utils/PipelineTestEnvironment.java
@@ -251,7 +251,7 @@ private static Version parseVersion(String version) {
                 versionParts.get(0), versionParts.get(1), versionParts.get(2), null, null, null);
     }
 
-    private static String getFlinkProperties(String flinkVersion) {
+    protected static String getFlinkProperties(String flinkVersion) {
         // this is needed for oracle-cdc tests.
         // see https://stackoverflow.com/a/47062742/4915129
         String javaOptsConfig;