[Fix][Zeta] Propagate interrupted pending job insertion

seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/CoordinatorService.java

@@ -1155,7 +1155,15 @@ private void restoreJobFromMasterActiveSwitch(@NonNull Long jobId, @NonNull JobI
         }
 
         PendingJobInfo pendingJobInfo = new PendingJobInfo(PendingSourceState.RESTORE, jobMaster);
-        pendingJobQueue.put(pendingJobInfo);
+        try {
+            pendingJobQueue.put(pendingJobInfo);
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            throw new SeaTunnelEngineException(
+                    String.format(
+                            "Job id %s restore interrupted while entering pending queue", jobId),
+                    e);
+        }
         jobMaster.getPhysicalPlan().updateJobState(JobStatus.PENDING);
         logger.info(String.format("The restore job enter pending queue, JobId: %s", jobId));
     }

seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/utils/PeekBlockingQueue.java

@@ -17,10 +17,6 @@
 
 package org.apache.seatunnel.engine.server.utils;
 
-import org.apache.seatunnel.common.utils.ExceptionUtils;
-
-import lombok.extern.slf4j.Slf4j;
-
 import java.util.Map;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ConcurrentHashMap;
@@ -42,7 +38,6 @@
  * 2. Check if resources are sufficient. <br>
  * 3. If resources are sufficient, take() the data; otherwise, do not take data from the queue.
  */
-@Slf4j
 public class PeekBlockingQueue<E> {
 
     private final BlockingQueue<E> queue = new LinkedBlockingQueue<>();
@@ -56,15 +51,16 @@ public PeekBlockingQueue(Function<E, Long> idExtractor) {
         this.idExtractor = idExtractor;
     }
 
-    public void put(E element) {
+    public void put(E element) throws InterruptedException {
         lock.lock();
         try {
             queue.put(element);
             Long jobId = idExtractor.apply(element);
             jobIdMap.put(jobId, element);
             notEmpty.signalAll();
         } catch (InterruptedException e) {
-            log.error("Put element into queue failed. {}", ExceptionUtils.getMessage(e));
+            Thread.currentThread().interrupt();
+            throw e;
         } finally {
             lock.unlock();
         }

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

@@ -57,6 +57,7 @@
 import org.apache.seatunnel.engine.server.resourcemanager.resource.SlotProfile;
 import org.apache.seatunnel.engine.server.task.operation.ReportMetricsOperation;
 import org.apache.seatunnel.engine.server.utils.NodeEngineUtil;
+import org.apache.seatunnel.engine.server.utils.PeekBlockingQueue;
 
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Disabled;
@@ -75,6 +76,7 @@
 import lombok.extern.slf4j.Slf4j;
 
 import java.lang.reflect.Field;
+import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
 import java.nio.file.Path;
 import java.nio.file.Paths;
@@ -802,9 +804,14 @@ private JobMaster enqueueMockPendingJob(
                 .when(jobMaster)
                 .run();
 
-        coordinatorService
-                .getPendingJobQueue()
-                .put(new PendingJobInfo(PendingSourceState.SUBMIT, jobMaster));
+        try {
+            coordinatorService
+                    .getPendingJobQueue()
+                    .put(new PendingJobInfo(PendingSourceState.SUBMIT, jobMaster));
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            throw new AssertionError("Failed to enqueue mock pending job", e);
+        }
         return jobMaster;
     }
 
@@ -1330,6 +1337,60 @@ void testSubmitJobOperationCanCompleteOnHazelcastOperationThread() {
         }
     }
 
+    @Test
+    @SetEnvironmentVariable(key = SKIP_CHECK_JAR, value = "true")
+    void testInterruptedPendingJobInsertionFailsSubmission() throws Exception {
+        String clusterName =
+                TestUtils.getClusterName(
+                        "CoordinatorServiceTest_testInterruptedPendingJobInsertionFailsSubmission");
+        HazelcastInstanceImpl instance =
+                SeaTunnelServerStarter.createHazelcastInstance(clusterName);
+        try {
+            SeaTunnelServer server =
+                    instance.node.getNodeEngine().getService(SeaTunnelServer.SERVICE_NAME);
+            CoordinatorService coordinatorService = server.getCoordinatorService();
+            InterruptiblePendingJobQueue pendingJobQueue = new InterruptiblePendingJobQueue();
+            ReflectionUtils.setField(coordinatorService, "pendingJobQueue", pendingJobQueue);
+
+            long jobId = instance.getFlakeIdGenerator(Constant.SEATUNNEL_ID_GENERATOR_NAME).newId();
+            LogicalDag logicalDag =
+                    TestUtils.createTestLogicalPlan(
+                            "batch_fake_to_console.conf",
+                            "interrupted_pending_job_insertion",
+                            jobId);
+            JobImmutableInformation jobImmutableInformation =
+                    new JobImmutableInformation(
+                            jobId,
+                            "Test",
+                            instance.getSerializationService(),
+                            logicalDag,
+                            Collections.emptyList(),
+                            Collections.emptyList());
+            Data data = instance.getSerializationService().toData(jobImmutableInformation);
+
+            PassiveCompletableFuture<Void> submitFuture =
+                    coordinatorService.submitJob(
+                            jobId, data, jobImmutableInformation.isStartWithSavePoint());
+            Assertions.assertTrue(
+                    pendingJobQueue.putStarted.await(20, TimeUnit.SECONDS),
+                    "submission did not reach pending queue insertion");
+
+            coordinatorService.clearCoordinatorService();
+
+            await().atMost(20, TimeUnit.SECONDS)
+                    .untilAsserted(
+                            () -> Assertions.assertTrue(submitFuture.isCompletedExceptionally()));
+            Assertions.assertThrows(CompletionException.class, submitFuture::join);
+            Assertions.assertFalse(pendingJobQueue.contains(jobId));
+            Assertions.assertNotEquals(
+                    JobStatus.PENDING,
+                    instance.getMap(Constant.IMAP_RUNNING_JOB_STATE).get(jobId),
+                    "an interrupted submission must not advance the job to PENDING");
+        } finally {
+            instance.shutdown();
+        }
+    }
+
     @Test
     void testGetPendingJobInfo() {
         SeaTunnelServer server = Mockito.mock(SeaTunnelServer.class);
@@ -1787,6 +1848,61 @@ void testRestoreUsesProvidedJobInfoInitializationTimestamp() throws Exception {
         }
     }
 
+    @Test
+    void testInterruptedPendingJobInsertionDuringRestoreFailsRestore() throws Exception {
+        HazelcastInstanceImpl instance =
+                createHazelcastInstanceWithJoinPortTryCount(
+                        TestUtils.getClusterName(
+                                "CoordinatorServiceTest_testInterruptedPendingJobInsertionDuringRestore"),
+                        1);
+        try {
+            SeaTunnelServer server =
+                    instance.node.getNodeEngine().getService(SeaTunnelServer.SERVICE_NAME);
+            CoordinatorService coordinatorService = server.getCoordinatorService();
+            await().atMost(60, TimeUnit.SECONDS)
+                    .untilAsserted(
+                            () -> Assertions.assertTrue(coordinatorService.isCoordinatorActive()));
+
+            long jobId = instance.getFlakeIdGenerator(Constant.SEATUNNEL_ID_GENERATOR_NAME).newId();
+            LogicalDag logicalDag =
+                    TestUtils.createTestLogicalPlan(
+                            "stream_fake_to_console.conf", "interrupted_restore", jobId);
+            JobImmutableInformation jobImmutableInformation =
+                    new JobImmutableInformation(
+                            jobId,
+                            "Test",
+                            instance.getSerializationService(),
+                            logicalDag,
+                            Collections.emptyList(),
+                            Collections.emptyList());
+            JobInfo jobInfo =
+                    new JobInfo(
+                            100L,
+                            instance.getSerializationService().toData(jobImmutableInformation));
+            IMap<Object, Object> runningJobStateIMap =
+                    instance.getMap(Constant.IMAP_RUNNING_JOB_STATE);
+            runningJobStateIMap.put(jobId, JobStatus.RUNNING);
+            ReflectionUtils.setField(
+                    coordinatorService, "pendingJobQueue", new AlwaysInterruptedPendingJobQueue());
+
+            InvocationTargetException invocationException =
+                    Assertions.assertThrows(
+                            InvocationTargetException.class,
+                            () ->
+                                    invokeRestoreJobFromMasterActiveSwitch(
+                                            coordinatorService, jobId, jobInfo));
+            Assertions.assertInstanceOf(
+                    SeaTunnelEngineException.class, invocationException.getCause());
+            Assertions.assertInstanceOf(
+                    InterruptedException.class, invocationException.getCause().getCause());
+            Assertions.assertFalse(coordinatorService.getPendingJobQueue().contains(jobId));
+            Assertions.assertNotEquals(JobStatus.PENDING, runningJobStateIMap.get(jobId));
+        } finally {
+            Thread.interrupted();
+            instance.shutdown();
+        }
+    }
+
     @Test
     @Disabled("Disabled because we can't know when the master node switches in the unit tests")
     void testJobRestoreWhenMasterNodeSwitch() {
@@ -2008,6 +2124,38 @@ private double runOps(
         return elapsedNs / 1_000_000_000.0;
     }
 
+    private static class InterruptiblePendingJobQueue extends PeekBlockingQueue<PendingJobInfo> {
+        private final CountDownLatch putStarted = new CountDownLatch(1);
+        private final CountDownLatch releasePut = new CountDownLatch(1);
+
+        private InterruptiblePendingJobQueue() {
+            super(PendingJobInfo::getJobId);
+        }
+
+        @Override
+        public void put(PendingJobInfo element) throws InterruptedException {
+            putStarted.countDown();
+            try {
+                releasePut.await();
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+            }
+            super.put(element);
+        }
+    }
+
+    private static class AlwaysInterruptedPendingJobQueue
+            extends PeekBlockingQueue<PendingJobInfo> {
+        private AlwaysInterruptedPendingJobQueue() {
+            super(PendingJobInfo::getJobId);
+        }
+
+        @Override
+        public void put(PendingJobInfo element) throws InterruptedException {
+            throw new InterruptedException("test interruption");
+        }
+    }
+
     private static class JobInformation {
 
         public final HazelcastInstanceImpl coordinatorServiceTest;

seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/utils/PeekBlockingQueueTest.java

@@ -115,7 +115,7 @@ public void testMultiPeekBlocking() throws InterruptedException, ExecutionExcept
     }
 
     @Test
-    public void testClear() {
+    public void testClear() throws InterruptedException {
         queue.put("1");
         queue.put("2");
         queue.put("3");