Checkpoint 58 - Remove unnecessary changes

Author: voonhous

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-hudi/src/main/java/org/apache/flink/cdc/connectors/hudi/sink/coordinator/MultiTableStreamWriteOperatorCoordinator.java

@@ -565,7 +565,8 @@ private void handleEnhancedWriteMetadataEvent(EnhancedWriteMetadataEvent enhance
         TableId tableId = pathToTableId.get(tablePath);
 
         if (tableId == null) {
-            LOG.warn("No tableId found for path: {}. Cannot process event.", tablePath);
+            LOG.error("No tableId found for path: {}. Cannot process event.", tablePath);
+            context.failJob(new IllegalStateException("No tableId found for path: " + tablePath));
             return;
         }
 

flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/MySqlToHudiE2eITCase.java

@@ -812,11 +812,6 @@ public void waitUntilJobState(JobID jobId, Duration timeout, JobStatus expectedS
             JobStatusMessage message = optMessage.get();
             JobStatus jobStatus = message.getJobState();
             if (!expectedStatus.isTerminalState() && jobStatus.isTerminalState()) {
-                try {
-                    Thread.sleep(50000);
-                } catch (InterruptedException e) {
-                    throw new RuntimeException(e);
-                }
                 throw new ValidationException(
                         String.format(
                                 "Job has been terminated! JobName: %s, JobID: %s, Status: %s",

flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/MySqlToIcebergE2eITCase.java

@@ -243,7 +243,6 @@ public void testSyncWholeDatabase() throws Exception {
             LOG.error("Update table for CDC failed.", e);
             throw e;
         }
-
         List<String> recordsInSnapshotPhase =
                 new ArrayList<>(
                         Arrays.asList(
@@ -262,7 +261,6 @@ public void testSyncWholeDatabase() throws Exception {
         recordsInSnapshotPhase =
                 recordsInSnapshotPhase.stream().sorted().collect(Collectors.toList());
         validateSinkResult(warehouse, database, "products", recordsInSnapshotPhase);
-        Thread.sleep(3600000L);
     }
 
     /**

flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/utils/PipelineTestEnvironment.java

@@ -135,7 +135,7 @@ private int getParallelism() {
                     "blob.server.port: 6124",
                     "taskmanager.numberOfTaskSlots: 10",
                     "parallelism.default: 4",
-                    "execution.checkpointing.interval: 30s",
+                    "execution.checkpointing.interval: 300",
                     "state.backend.type: hashmap",
                     "env.java.opts.all: -Doracle.jdbc.timezoneAsRegion=false",
                     "execution.checkpointing.savepoint-dir: file:///opt/flink",
@@ -389,11 +389,6 @@ public void waitUntilJobState(Duration timeout, JobStatus expectedStatus) {
                 JobStatusMessage message = jobStatusMessages.iterator().next();
                 JobStatus jobStatus = message.getJobState();
                 if (!expectedStatus.isTerminalState() && jobStatus.isTerminalState()) {
-                    try {
-                        Thread.sleep(50000);
-                    } catch (InterruptedException e) {
-                        throw new RuntimeException(e);
-                    }
                     throw new ValidationException(
                             String.format(
                                     "Job has been terminated! JobName: %s, JobID: %s, Status: %s",