[Fix][Zeta] Fix pendingJobMasterMap resource leak

Author: hawk9821

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCWithBinlogDeleteIT.java

@@ -208,15 +208,8 @@ public void testRestoreTaskWhenBinlogDelete(TestContainer container)
                 .atMost(60, TimeUnit.SECONDS)
                 .untilAsserted(
                         () ->
-                                Assertions.assertNotEquals(
-                                        "FAILED", container.getJobStatus(String.valueOf(jobId))));
-
-        // check no error
-        log.info("****************** container logs start ******************");
-        String containerLogs = container.getServerLogs();
-        log.info(containerLogs);
-        Assertions.assertFalse(containerLogs.contains("ERROR"));
-        log.info("****************** container logs end ******************");
+                                Assertions.assertEquals(
+                                        "RUNNING", container.getJobStatus(String.valueOf(jobId))));
 
         // cancel task
         Assertions.assertEquals(0, container.cancelJob(String.valueOf(jobId)).getExitCode());

seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/JobExecutionIT.java

@@ -119,11 +119,7 @@ public void cancelJobTest() throws Exception {
                     engineClient.createExecutionContext(filePath, jobConfig, SEATUNNEL_CONFIG);
 
             final ClientJobProxy clientJobProxy = jobExecutionEnv.execute();
-            await().atMost(30000, TimeUnit.MILLISECONDS)
-                    .untilAsserted(
-                            () ->
-                                    Assertions.assertFalse(
-                                            clientJobProxy.getJobStatus().isEndState()));
+            Assertions.assertFalse(clientJobProxy.getJobStatus().isEndState());
 
             CompletableFuture<JobStatus> objectCompletableFuture =
                     CompletableFuture.supplyAsync(clientJobProxy::waitForJobComplete);