[Fix][Zeta] Fix pendingJobMasterMap resource leak

Author: hawk9821

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

@@ -619,9 +619,6 @@ public void testTwoPipelineBatchJobRestoreIn2NodeMasterDown() throws Exception {
                                                 + FileUtils.getFileLineNumberFromDir(
                                                         testResources.getLeft())
                                                 + "=================================\n");
-                                JobStatus jobStatus = clientJobProxy.getJobStatus();
-                                System.out.println("++++++++++++++++++++++++++++++++++++++++");
-                                System.out.println(jobStatus);
                                 Assertions.assertTrue(objectCompletableFuture.isDone());
                                 Assertions.assertEquals(
                                         JobStatus.FINISHED, objectCompletableFuture.get());

seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/CoordinatorServiceTest.java

@@ -95,8 +95,8 @@ public void testMasterNodeActive() {
                                 CoordinatorService coordinatorService =
                                         server2.getCoordinatorService();
                                 Assertions.assertTrue(coordinatorService.isCoordinatorActive());
-                            } catch (Exception e) {
-                                Assertions.assertInstanceOf(SeaTunnelEngineException.class, e);
+                            } catch (SeaTunnelEngineException e) {
+                                Assertions.fail("Should not throw SeaTunnelEngineException here.");
                             }
                         });
         instance2.shutdown();
@@ -203,6 +203,8 @@ void testCleanupRunningJobStateIMap() {
                         "batch_fake_to_console.conf",
                         "test_cleanup_running_job_state_imap");
         CoordinatorService coordinatorService = jobInformation.coordinatorService;
+        IMap<Object, Object> runningJobStateIMap =
+                coordinatorService.getJobMaster(jobInformation.jobId).getRunningJobStateIMap();
 
         await().atMost(10000, TimeUnit.MILLISECONDS)
                 .untilAsserted(
@@ -229,6 +231,7 @@ void testCleanupRunningJobStateIMap() {
                                     coordinatorService.getJobMaster(jobInformation.jobId);
                             // job master should be null
                             Assertions.assertNull(jobMaster);
+                            Assertions.assertTrue(runningJobStateIMap.isEmpty());
                         });
 
         jobInformation.coordinatorService.clearCoordinatorService();
@@ -384,13 +387,9 @@ private JobInformation submitJob(String testClassName, String jobConfigFile, Str
         Data data =
                 coordinatorServiceTest.getSerializationService().toData(jobImmutableInformation);
 
-        try {
-            coordinatorService
-                    .submitJob(jobId, data, jobImmutableInformation.isStartWithSavePoint())
-                    .join();
-        } catch (Throwable e) {
-            log.error("submit job failed", e);
-        }
+        coordinatorService
+                .submitJob(jobId, data, jobImmutableInformation.isStartWithSavePoint())
+                .join();
         return new JobInformation(coordinatorServiceTest, coordinatorService, jobId);
     }
 

seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/CoordinatorServiceWithCancelPendingJobTest.java

@@ -48,7 +48,6 @@
 import java.util.Collections;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicReference;
 
 import static org.awaitility.Awaitility.await;
 
@@ -170,28 +169,16 @@ private JobMaster newJobInstanceWithRunningState(long jobId, boolean restore)
                         .submitJob(jobId, data, jobImmutableInformation.isStartWithSavePoint());
         voidPassiveCompletableFuture.join();
 
-        AtomicReference<JobMaster> jobMasterReference = new AtomicReference<>();
-
-        await().atMost(30000, TimeUnit.MILLISECONDS)
-                .untilAsserted(
-                        () -> {
-                            JobMaster jobMaster =
-                                    server.getCoordinatorService().getJobMaster(jobId);
-                            Assertions.assertNotNull(jobMaster);
-                            jobMasterReference.set(jobMaster);
-                        });
+        JobMaster jobMaster = server.getCoordinatorService().getJobMaster(jobId);
 
         // waiting for job status turn to running
         await().atMost(120, TimeUnit.SECONDS)
                 .untilAsserted(
-                        () ->
-                                Assertions.assertEquals(
-                                        JobStatus.PENDING,
-                                        jobMasterReference.get().getJobStatus()));
+                        () -> Assertions.assertEquals(JobStatus.PENDING, jobMaster.getJobStatus()));
 
         // Because handleCheckpointTimeout is an async method, so we need sleep 5s to waiting job
         // status become running again
         Thread.sleep(5000);
-        return jobMasterReference.get();
+        return jobMaster;
     }
 }

seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/checkpoint/CheckpointErrorRestoreEndTest.java

@@ -27,7 +27,6 @@
 import org.junit.jupiter.api.condition.OS;
 
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicReference;
 
 import static org.awaitility.Awaitility.await;
 
@@ -42,25 +41,14 @@ public void testCheckpointRestoreToFailEnd() {
         long jobId = System.currentTimeMillis();
         startJob(jobId, STREAM_CONF_WITH_ERROR_PATH, false);
 
-        AtomicReference<JobMaster> jobMasterReference = new AtomicReference<>();
-        await().atMost(60, TimeUnit.SECONDS)
-                .untilAsserted(
-                        () -> {
-                            JobMaster jobMaster =
-                                    server.getCoordinatorService().getJobMaster(jobId);
-                            Assertions.assertNotNull(jobMaster);
-                            jobMasterReference.set(jobMaster);
-                        });
-
-        Assertions.assertEquals(
-                1, jobMasterReference.get().getPhysicalPlan().getPipelineList().size());
+        JobMaster jobMaster = server.getCoordinatorService().getJobMaster(jobId);
+        Assertions.assertEquals(1, jobMaster.getPhysicalPlan().getPipelineList().size());
         await().atMost(240, TimeUnit.SECONDS)
                 .untilAsserted(
                         () ->
                                 Assertions.assertEquals(
                                         3,
-                                        jobMasterReference
-                                                .get()
+                                        jobMaster
                                                 .getPhysicalPlan()
                                                 .getPipelineList()
                                                 .get(0)

seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/master/JobMasterTest.java

@@ -51,7 +51,6 @@
 import java.util.Collections;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicReference;
 
 import static org.awaitility.Awaitility.await;
 
@@ -311,28 +310,16 @@ private JobMaster newJobInstanceWithRunningState(long jobId, boolean restore)
                         .submitJob(jobId, data, jobImmutableInformation.isStartWithSavePoint());
         voidPassiveCompletableFuture.join();
 
-        AtomicReference<JobMaster> jobMasterReference = new AtomicReference<>();
-
-        await().atMost(30000, TimeUnit.MILLISECONDS)
-                .untilAsserted(
-                        () -> {
-                            JobMaster jobMaster =
-                                    server.getCoordinatorService().getJobMaster(jobId);
-                            Assertions.assertNotNull(jobMaster);
-                            jobMasterReference.set(jobMaster);
-                        });
+        JobMaster jobMaster = server.getCoordinatorService().getJobMaster(jobId);
 
         // waiting for job status turn to running
         await().atMost(120000, TimeUnit.MILLISECONDS)
                 .untilAsserted(
-                        () ->
-                                Assertions.assertEquals(
-                                        JobStatus.RUNNING,
-                                        jobMasterReference.get().getJobStatus()));
+                        () -> Assertions.assertEquals(JobStatus.RUNNING, jobMaster.getJobStatus()));
 
         // Because handleCheckpointTimeout is an async method, so we need sleep 5s to waiting job
         // status become running again
         Thread.sleep(5000);
-        return jobMasterReference.get();
+        return jobMaster;
     }
 }