[Fix][Zeta] Propagate interrupted pending job insertion

[CryoThrust]

Purpose

Fixes #12010. An interrupted pending-job insertion was previously swallowed by PeekBlockingQueue.put, allowing submitJob to complete successfully even though the job was not queued.

Changes

• Propagate InterruptedException from PeekBlockingQueue.put and restore the interrupt flag.
• Let CoordinatorService.submitJob complete its future exceptionally through the existing error path and avoid advancing the physical plan to PENDING.
• Apply the same explicit interruption handling to the master-switch restore path.
• Add a deterministic regression test covering queue state, future completion, and job status.
• Audited take and peekBlocking; they already propagate interruption.

Verification

• git diff --check passes.
• Focused Maven compile/test could not start because the configured internal repository reset the connection and the HTTP plugin mirror was blocked by Maven.

[DanielLeens]

What Problem Does This PR Solve?

Today, if the coordinator thread that is inserting a job into PeekBlockingQueue (the pending-job queue) gets interrupted — most concretely by CoordinatorService.clearCoordinatorService() calling executorService.shutdownNow() during a master step-down/failover — PeekBlockingQueue.put() silently swallows the InterruptedException, logs one line, and returns normally. CoordinatorService then proceeds as if the insert succeeded: it advances the physical plan to PENDING, logs "enter the pending queue", and (for submitJob) completes the client's submit future successfully, even though the job was never actually added to the queue. The job is effectively lost from the local scheduling view while everyone downstream believes the submission worked.

This PR (fixing #12010) makes PeekBlockingQueue.put() propagate InterruptedException (restoring the interrupt flag first) instead of swallowing it, and updates both call sites in CoordinatorService (submitJob and restoreJobFromMasterActiveSwitch) to stop advancing job state / reporting success when the insert didn't actually happen.

1. Code Change Review

1.1 Core Logic Analysis

Files touched:

• seatunnel-engine/seatunnel-engine-server/.../utils/PeekBlockingQueue.java
• seatunnel-engine/seatunnel-engine-server/.../CoordinatorService.java
• seatunnel-engine/seatunnel-engine-server/.../CoordinatorServiceTest.java

PeekBlockingQueue.put() (utils/PeekBlockingQueue.java:54-67): signature changed from void put(E) to void put(E) throws InterruptedException. On interrupt it now does Thread.currentThread().interrupt(); throw e; instead of log.error(...) and swallowing. This is the textbook-correct pattern for handling InterruptedException you don't want to fully absorb.

One simple before/after example: a client calls submitJob(jobId, ...). Concurrently this node loses master status and clearCoordinatorService() runs, calling executorService.shutdownNow(), which interrupts the in-flight submit task right as it calls pendingJobQueue.put(pendingJobInfo).

• Before: put() swallows the interrupt, the submit task continues, jobSubmitFuture.complete(null) — client sees success — but pendingJobQueue never contains the job, so nothing will ever schedule it locally.
• After: put() throws, submitJob's catch (Throwable e) completes the future exceptionally, runningJobInfoIMap/runningJobMasterMap/pendingJobQueue are cleaned up for that jobId, and the physical plan is never advanced to PENDING. The client gets an honest failure instead of a false success.

Call chain / does the normal path reach this? pendingJobQueue.put() sits on the hot path of every single job submission (CoordinatorService.java:1458, inside submitJob's async task) and every master-switch restore (CoordinatorService.java:1159, inside restoreJobFromMasterActiveSwitch), so the code path is exercised on every normal submission. However, the new throw-instead-of-swallow behavior is only observable on the boundary/race condition where the executor is interrupted mid-insert — in practice this means a clearCoordinatorService()/shutdownNow() racing with an in-flight submit or restore (confirmed at CoordinatorService.java:1320: executorService.shutdownNow()). So: normal-path code is touched, but the behavioral delta only fires on a recovery/failover race window, which matches the PR's own framing.

Verifying the claimed problem is real (step 5 of the review protocol): I read PeekBlockingQueue.put() pre-fix and the linked issue #12010, which includes a deterministic reproduction (interrupt injected at the exact moment before the real queue.put(), driven by the real clearCoordinatorService()/shutdownNow() path, not a test-only shortcut). The new regression test added by this PR (CoordinatorServiceTest.java) reproduces the same scenario against the real CoordinatorService.submitJob() + clearCoordinatorService() machinery and passes with the fix, which is solid evidence the fix addresses the real bug and not just a symptom.

I also checked both put() call sites for the checked-exception ripple effect (grep -rn "PeekBlockingQueue" seatunnel-engine/seatunnel-engine-server/src, and confirmed no other module references PeekBlockingQueue, so this is fully contained inside seatunnel-engine-server):

• CoordinatorService.java:1458 (submitJob, normal path) is inside try { ... } catch (Throwable e) { ... } (CoordinatorService.java:1461), so it already compiles against the new checked exception and the existing broad catch reports failure via jobSubmitFuture.completeExceptionally(...) with a full stack trace (ExceptionUtils.getMessage(e) prints the whole trace, not just getMessage(), so the interrupted case is still diagnosable even without a custom message).
• CoordinatorService.java:1157-1165 (restoreJobFromMasterActiveSwitch) is given an explicit catch (InterruptedException e) with a jobId-scoped message ("Job id %s restore interrupted while entering pending queue"), wrapped in SeaTunnelEngineException, which propagates up and is logged once at CoordinatorService.java:1074 (logger.severe(e)) inside restoreAllRunningJobFromMasterNodeSwitch's per-job runAsync — no swallowing, no duplicate logging.

1.2 Compatibility Impact

Fully compatible.

• PeekBlockingQueue is an internal seatunnel-engine-server utility with no consumers outside that module (verified by grep across the repo).
• No config Option, public API, SPI, checkpoint/savepoint format, or serialization protocol is touched.
• The externally-visible behavior change is that a submission that races with a master step-down and previously reported false success now correctly reports failure. Since the old behavior was itself the bug being fixed (job silently missing from the pending queue despite a "success" response), this is a correctness fix, not a backward-incompatible behavior change that needs a migration note in docs/en/introduction/concepts/incompatible-changes.md.

1.3 Performance / Side-Effect Analysis

No new locks, no new polling loops, no additional allocations on the hot path beyond the try/catch itself (negligible). Thread.currentThread().interrupt() is called exactly once per failure and is idempotent if called again upstream (the restore path calls it a second time at CoordinatorService.java:1161, which is harmless — setting an already-set interrupt flag is a no-op).

One thing I traced carefully because it looked like a possible resource leak at first: on the interrupted-put path, the JobMaster that was already constructed and init()-ed just before the failed put() call (CoordinatorService.java:1136-1152 for restore, 1426-1452 for submit) is discarded without an explicit jobMaster.interrupt(). At that moment it is not yet present in runningJobMasterMap or pendingJobQueue, so clearCoordinatorService()'s interrupt sweep (CoordinatorService.java:1295-1318) cannot reach it either. I read JobMaster.init() → initCheckPointManager() (JobMaster.java:230-330, 332-348) to check whether this leaves anything live behind: initCheckPointManager() only constructs a CheckpointManager object, it does not start any scheduled threads or timers at that point (those only start once the job is actually dequeued and physicalPlan.startJob() runs). So this is not a live-thread/resource leak — the half-initialized JobMaster just becomes normal garbage once the local reference goes out of scope. Not a blocker, just noting it so nobody has to re-derive this if the issue's proposed "explicitly interrupt the initialized JobMaster" contract item is picked up later.

1.4 Error Handling and Logging

• No swallowed exceptions remain in PeekBlockingQueue.put() — the log statement was removed together with the now-unused @Slf4j/ExceptionUtils imports, and the exception is instead logged exactly once by each caller (no duplicate/missing logging).
• No sensitive data is logged.
• Minor asymmetry (non-blocking): the restore path gets an explicit catch (InterruptedException e) with a descriptive, jobId-scoped message, while the submit path relies on the generic catch (Throwable e). Functionally both are diagnosable (the generic path still logs a full stack trace via ExceptionUtils.getMessage), but giving submitJob the same explicit, jobId-scoped message as the restore path would make the two call sites of the same underlying condition consistent and slightly easier to grep in production logs. Optional polish, not required for merge.

2. Code Quality Assessment

2.1 Coding Standards

Clean, minimal diff. Removing the now-dead @Slf4j annotation and ExceptionUtils import from PeekBlockingQueue.java alongside the log-statement removal is good hygiene — no leftover unused imports.

2.2 Test Coverage and Test Stability

This is the actual blocker for this PR, and it's a compile break, not a logic bug.

PeekBlockingQueue.put()'s new throws InterruptedException also affects PeekBlockingQueueTest.java, an existing test file that this PR's diff does not touch. Three of its four test methods already declare throws InterruptedException (testBasic, testPeekBlocking, testMultiPeekBlocking), but testClear() (PeekBlockingQueueTest.java:117-125) does not, and calls queue.put(...) three times (lines 119-121) without declaring or catching the new checked exception.

I independently verified this against the fork's own CI run (apache/seatunnel's "Build" check is only a pointer to the contributor-fork Actions run — dereferenced via repos/CryoThrust/seatunnel/actions/runs?head_sha=<head> to run 33717480166). The seatunnel-engine-server testCompile goal fails with:

[ERROR] .../PeekBlockingQueueTest.java:[119,17] error: unreported exception InterruptedException; must be caught or declared to be thrown
[ERROR] .../PeekBlockingQueueTest.java:[120,17] error: unreported exception InterruptedException; must be caught or declared to be thrown
[ERROR] .../PeekBlockingQueueTest.java:[121,17] error: unreported exception InterruptedException; must be caught or declared to be thrown
[ERROR] Failed to execute goal ...maven-compiler-plugin:3.10.1:testCompile ... on project seatunnel-engine-server

This single compile failure in seatunnel-engine-server is why essentially every job in the fork's Build workflow is red (unit-test x4, and every IT module that depends on seatunnel-engine-server for reactor build order) — it is not flaky/pre-existing/environmental; it is a direct, deterministic consequence of this PR's own signature change against a file the PR didn't update. This matches the PR author's own "Verification" note that a focused Maven build "could not start" locally — the break was never actually compiled locally before pushing.

Fix: add throws InterruptedException to testClear()'s signature (consistent with the other three methods in the same file), or catch it locally if the intent is to keep that one test interruption-agnostic.

Stability rating for the new regression test (CoordinatorServiceTest.testInterruptedPendingJobInsertionFailsSubmission, added at CoordinatorServiceTest.java:1339+, using the InterruptiblePendingJobQueue helper at the bottom of the file): Stable. It synchronizes purely via CountDownLatch (putStarted/releasePut) plus Awaitility.await().atMost(...).untilAsserted(...) — no Thread.sleep-based race, no arbitrary timing assumption. It drives the interrupt through the real production trigger (coordinatorService.clearCoordinatorService() → executorService.shutdownNow()), not a synthetic Thread.interrupt() call on a test-owned thread, so it's a faithful reproduction of the actual race rather than a shortcut. It asserts all three relevant postconditions: future fails exceptionally, the queue does not contain the job, and the job state map is not advanced to PENDING.

Coverage gap (non-blocking, recommended): the PR modifies two call sites (submitJob and restoreJobFromMasterActiveSwitch), but only submitJob gets a dedicated regression test. restoreJobFromMasterActiveSwitch's new explicit catch (InterruptedException e) branch (CoordinatorService.java:1160-1165) is currently only exercised implicitly, if at all. Given the PR description explicitly calls out "Apply the same explicit interruption handling to the master-switch restore path" as one of its changes, a second test mirroring testInterruptedPendingJobInsertionFailsSubmission but for the restore path (interrupting restoreJobFromMasterActiveSwitch's pendingJobQueue.put() and asserting it throws SeaTunnelEngineException with the expected message, and that state isn't advanced) would close the loop and guard against future regressions on that path too.

2.3 Documentation Updates

Not required. This is a purely internal engine-server correctness fix — no new/changed config Option, CLI flag, or public-facing behavior contract, so no docs/en/docs/zh update is needed.

3. Architectural Soundness

3.1 Elegance of the Solution

The "propagate instead of swallow" fix is the simplest correct fix for the reported problem and matches option 1 of the two contracts the author laid out in issue #12010 (fail the interrupted/stale submission and preserve the interrupt flag, vs. making acceptance fully non-interruptible/epoch-coordinated). Worth a note for the maintainers: the issue explicitly says "Are you willing to submit a PR? Yes, after maintainers confirm the intended interruption contract," but I don't see a maintainer confirmation recorded before this PR was opened implementing contract option 1. That's a project-process question for the maintainers to weigh in on, not a code defect — option 1 is a safe, minimal, and correct choice on its own merits (it just doesn't attempt the more invasive epoch-coordinated non-interruptible-insert design from option 2, which would be a much larger change).

3.2 Maintainability

Good — the two call sites now make the interrupted-insert case an explicit, visible branch instead of a silent no-op buried inside a utility class, which is easier for future readers to reason about.

3.3 Extensibility

No concerns; this doesn't add new abstractions or contracts beyond what already existed.

3.4 Historical-Version Compatibility

No serialization, checkpoint/savepoint, or config format is touched. Nothing here affects upgrade paths from previously released versions.

4. Issue Summary

#	Issue	Location	Severity
1	testClear() doesn't declare/catch the new checked InterruptedException from PeekBlockingQueue.put(), breaking seatunnel-engine-server test-compile and cascading to essentially every CI job (verified against the fork's own Build run, not the apache-side pointer check)	seatunnel-engine-server/src/test/java/.../utils/PeekBlockingQueueTest.java:117-125	High (blocking, but a one-line/trivial fix)
2	restoreJobFromMasterActiveSwitch's new explicit interrupted-put branch has no dedicated regression test (only the submitJob path is covered)	CoordinatorService.java:1157-1165	Medium (recommended, non-blocking)
3	Minor asymmetry: submitJob relies on the generic catch (Throwable e) for the interrupted-put case instead of an explicit, jobId-scoped message like the restore path gets	CoordinatorService.java:1450-1465	Low (optional polish)

5. Merge Recommendation

Conclusion: Ready to merge after fixes

1. Blockers — must be fixed

   • Issue #1: fix the pre-existing PeekBlockingQueueTest.testClear() compile break introduced by the new checked-exception signature (add throws InterruptedException or catch it locally), then re-run CI so it goes green on the merits of the actual fix.

2. Recommended fixes — non-blocking

   • Issue #2: add a regression test for the restoreJobFromMasterActiveSwitch interrupted-put branch, mirroring the existing submitJob test, since the PR explicitly claims to cover both call sites.
   • Issue #3: optionally give submitJob the same explicit, jobId-scoped InterruptedException message the restore path already has, for log consistency.

Overall assessment: the root-cause diagnosis in #12010 is precise and well-evidenced, and the fix itself — propagate instead of swallow, restore the interrupt flag, stop advancing job state on failure — is the right, minimally invasive shape for this bug. The new submitJob regression test is genuinely well built (deterministic, drives the real production interrupt path, no sleep-based flakiness). The only thing standing between this PR and a green build is the missed pre-existing test caller, which is a very small, mechanical fix. Great first contribution — thank you for the thorough issue report and reproduction, and welcome to the SeaTunnel community! Looking forward to seeing this land once the build is green.

[DanielLeens]

What Problem Does This PR Solve?

If the coordinator thread that is inserting a job into PeekBlockingQueue (the pending-job queue) gets interrupted — most concretely by CoordinatorService.clearCoordinatorService() calling executorService.shutdownNow() during a master step-down/failover — PeekBlockingQueue.put() used to silently swallow the InterruptedException, log one line, and return normally. CoordinatorService then proceeded as if the insert had succeeded: it advanced the physical plan to PENDING and, for submitJob, completed the client's submit future successfully, even though the job was never actually added to the queue. The job was effectively lost from the local scheduling view while everyone downstream believed the submission worked.

This PR (fixing #12010) makes PeekBlockingQueue.put() propagate InterruptedException (restoring the interrupt flag first) instead of swallowing it, and updates both call sites in CoordinatorService (submitJob and restoreJobFromMasterActiveSwitch) to stop advancing job state / reporting success when the insert didn't actually happen.

This is a full re-review since a new set of commits landed after my previous pass. The head commit I reviewed before (df53e85b) is no longer reachable — the branch was rewritten/rebased and two new commits plus two formatting-only commits were added on top. I re-read the entire diff from scratch against current dev rather than trusting my earlier notes.

1. Code Change Review

1.1 Core Logic Analysis

Files touched: PeekBlockingQueue.java, CoordinatorService.java, CoordinatorServiceTest.java, PeekBlockingQueueTest.java.

PeekBlockingQueue.put() (utils/PeekBlockingQueue.java:54-66): signature changed from void put(E) to void put(E) throws InterruptedException. On interrupt it now does Thread.currentThread().interrupt(); throw e; (lines 62-63) instead of logging and swallowing. This is the textbook-correct pattern for an InterruptedException you don't want to fully absorb.

Call sites:

• CoordinatorService.java around the submitJob async task (pendingJobQueue.put(pendingJobInfo) inside the submit flow): this call was already wrapped in a broad try { ... } catch (Throwable e) { jobSubmitFuture.completeExceptionally(...); }, and the surrounding if (!jobSubmitFuture.isCompletedExceptionally()) gate that decides whether to advance the job to PENDING or clean up (runningJobInfoIMap.remove, runningJobMasterMap.remove, pendingJobQueue.removeById) was already present on dev before this PR. So submitJob itself needed no code change — it was already correctly structured to fail safely once put() actually throws. The only thing missing was put() throwing in the first place, which this PR supplies. I diffed this method against dev directly to confirm no lines changed here — that matches what the PR's diff stat shows (CoordinatorService.java changes are confined to the restore path).
• CoordinatorService.java:1159 (restoreJobFromMasterActiveSwitch) is the one call site that did need a code change: it's now wrapped in an explicit 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); }. This restores the interrupt flag, wraps with a jobId-scoped message, and propagates — no swallowing, and updateJobState(JobStatus.PENDING) is correctly skipped because the method throws before reaching that line.

I re-grepped for all usages of PeekBlockingQueue across the repo (grep -rn "PeekBlockingQueue" seatunnel-engine/) and confirmed it is fully contained inside seatunnel-engine-server, so the checked-exception signature change has no ripple effect outside this module.

Verifying the fix is real: the reproduction in issue #12010 injects the interrupt at the real trigger point (clearCoordinatorService() → executorService.shutdownNow()), not a synthetic shortcut, and the regression tests in this PR reproduce the same scenario against the real CoordinatorService.submitJob() / restoreJobFromMasterActiveSwitch() machinery. I traced both end-to-end below in 2.2 — this is solid evidence the fix targets the real bug, not a symptom.

1.2 Compatibility Impact

Fully compatible.

• PeekBlockingQueue is an internal seatunnel-engine-server utility with no consumers outside that module.
• No config Option, public API, SPI, checkpoint/savepoint format, or serialization protocol is touched.
• The externally-visible behavior change is that a submission or restore that races with a master step-down and previously reported false success now correctly reports failure. Since the old behavior was itself the bug (job silently missing from the pending queue despite a "success" response), this is a correctness fix, not a backward-incompatible change needing a migration note in docs/en/introduction/concepts/incompatible-changes.md.

1.3 Performance / Side-Effect Analysis

No new locks, no new polling loops, no additional allocations on the hot path beyond the try/catch itself. Thread.currentThread().interrupt() is idempotent if invoked twice (harmless if a caller further up also sets it).

As noted in my previous pass: on the interrupted-put path, the JobMaster already constructed and init()-ed just before the failed put() is discarded without an explicit interrupt sweep reaching it (it's not yet in runningJobMasterMap/pendingJobQueue at that point). I re-verified JobMaster.init() only constructs a CheckpointManager without starting scheduled threads/timers at that stage, so this remains not a live-thread/resource leak — just normal garbage once the reference goes out of scope. Not a blocker.

1.4 Error Handling and Logging

• No swallowed exceptions remain in PeekBlockingQueue.put(); the log statement, @Slf4j, and the now-unused ExceptionUtils import were correctly removed together.
• No sensitive data is logged.
• Minor asymmetry (still present, non-blocking, carried over from my last review): the restore path gets an explicit, jobId-scoped InterruptedException message, while submitJob still relies on the generic catch (Throwable e) path (still fully diagnosable via the logged stack trace, just less specific). Optional polish, not required for merge.

2. Code Quality Assessment

2.1 Coding Standards

Clean, minimal diff. Dead-import removal in PeekBlockingQueue.java is good hygiene.

2.2 Test Coverage and Test Stability

Both blockers from my previous review have been addressed in the code itself:

1. PeekBlockingQueueTest.testClear() (utils/PeekBlockingQueueTest.java:118) now declares throws InterruptedException, matching the other three test methods in the file and fixing the test-compile break I flagged last time.
2. A new regression test now covers the restoreJobFromMasterActiveSwitch path: testInterruptedPendingJobInsertionDuringRestoreFailsRestore (CoordinatorServiceTest.java:1852), using a new AlwaysInterruptedPendingJobQueue helper (CoordinatorServiceTest.java:2150) that unconditionally throws InterruptedException from put(). It swaps this into coordinatorService via ReflectionUtils.setField, invokes the private restoreJobFromMasterActiveSwitch via the existing reflection helper, and asserts the reflection call surfaces InvocationTargetException → SeaTunnelEngineException → InterruptedException, that the job never lands in the pending queue, and that job state never advances to PENDING. This is deterministic (no timing dependency at all — the mock always throws), and the finally { Thread.interrupted(); instance.shutdown(); } correctly clears the test thread's interrupt flag (set by the production code's Thread.currentThread().interrupt() call) before shutting down, so it can't bleed into other tests sharing the same thread.

Stability re-check of the existing submit-path test (testInterruptedPendingJobInsertionFailsSubmission, CoordinatorServiceTest.java:1342, via InterruptiblePendingJobQueue at CoordinatorServiceTest.java:2130): I traced this again end-to-end because interrupt-timing tests are exactly the kind of thing that looks stable on paper but races in practice. The mechanism is:

1. InterruptiblePendingJobQueue.put() counts down putStarted, then blocks on releasePut.await() (a real, deterministic block — no sleep).
2. The test waits on putStarted (bounded, 20s) before calling coordinatorService.clearCoordinatorService(), which calls executorService.shutdownNow() and interrupts the submit task's thread while it is genuinely parked inside releasePut.await().
3. The mock's catch (InterruptedException e) { Thread.currentThread().interrupt(); } resets the interrupt flag, then unconditionally calls super.put(element) — the real PeekBlockingQueue.put().
4. Because the thread's interrupt flag is now set, the inner LinkedBlockingQueue.put() call's putLock.lockInterruptibly() throws InterruptedException immediately (no blocking needed — an unbounded queue's lock acquisition doesn't contend, but lockInterruptibly() still checks interrupt status on entry), which is caught by the real PeekBlockingQueue.put() and rethrown per the fix.
5. That propagates out to submitJob's catch (Throwable e), completing jobSubmitFuture exceptionally.

Every step is driven by a CountDownLatch or a genuine lock-acquisition interrupt check, not Thread.sleep or an arbitrary timing assumption — I confirm this test is Stable, not flaky, and it faithfully reproduces the real production race rather than taking a shortcut.

Remaining coverage gap: none from my prior list — both call sites now have dedicated regression tests. I don't see any further coverage gap worth blocking on.

2.3 Documentation Updates

Not required — purely internal engine-server correctness fix, no config/API/behavior contract change.

3. Architectural Soundness

3.1 Elegance of the Solution

"Propagate instead of swallow" remains the simplest correct fix and matches contract option 1 from issue #12010 (fail the interrupted/stale submission and preserve the interrupt flag). Same note as before: the issue text says the author would submit a PR "after maintainers confirm the intended interruption contract" — I don't see a recorded maintainer confirmation, which is a process question for the maintainers, not a code defect.

3.2 Maintainability

Good — the interrupted-insert case is now an explicit, visible branch at both call sites instead of a silent no-op buried in a utility class.

3.3 Extensibility

No concerns; no new abstractions introduced.

3.4 Historical-Version Compatibility

No serialization, checkpoint/savepoint, or config format is touched. No impact on upgrade paths from previously released versions.

4. Issue Summary

#	Issue	Location	Severity	Raised by another reviewer
1	The final formatting commit hand-edits two blocks of CoordinatorServiceTest.java into a multi-line shape that the project's actual Spotless formatter (google-java-format, AOSP style) does not want — both lines fit within the 100-column limit and Spotless wants them collapsed back to a single line. I confirmed this against the fork's own live CI run for this exact head commit (515b0f73cb…): the Run / Code style job fails with spotless:check, reporting exactly these two hunks as violations, and every other job in the workflow (including unit-test, which would otherwise validate compilation of the new/changed test code) is gated behind it and shows skipped.	seatunnel-engine-server/src/test/java/.../CoordinatorServiceTest.java (the long jobId = ... and ReflectionUtils.setField(...) calls inside testInterruptedPendingJobInsertionDuringRestoreFailsRestore)	High (blocking, mechanical one-command fix)	No
2	Minor asymmetry: submitJob's interrupted-put case still relies on the generic catch (Throwable e) instead of an explicit, jobId-scoped message like the restore path has. Functionally both are diagnosable (a full stack trace is still logged), this is purely a log-grep-ability nicety.	CoordinatorService.java (submit-path catch block)	Low (optional polish)	No

5. Merge Recommendation

Conclusion: Ready to merge after fixes

1. Blockers — must be fixed

   • Issue #1: run the project's actual mvn spotless:apply (or equivalent IDE auto-format matching the repo's google-java-format/AOSP config) on CoordinatorServiceTest.java and push the result. The current head fails the fork's own Code style CI check, and because that check gates the rest of the workflow, unit-test — which is what would actually prove the new tests compile and pass — has not run yet on this commit. Once Spotless is green, the unit-test job needs to actually complete successfully before this can be considered CI-verified; right now that's an open item, not a confirmed pass.

2. Recommended fixes — non-blocking

   • Issue #2: optionally give submitJob the same explicit, jobId-scoped InterruptedException message the restore path already has, for log consistency.

Overall assessment: both substantive gaps from my previous review are now closed — the testClear() compile break is fixed exactly as suggested, and the restore path now has a dedicated, deterministic regression test mirroring the submit-path one. I re-traced both tests' interrupt-propagation mechanics end-to-end (not just read them) and both are genuinely stable, driven by real lock/latch semantics rather than sleeps or timing luck. The only thing standing between this PR and mergeable right now is a self-inflicted formatting regression in the very last commit — a one-command fix (mvn spotless:apply) followed by letting the full CI matrix actually run to completion. The underlying fix (propagate instead of swallow, restore the interrupt flag, don't advance job state on failure) is correct, minimal, and well-tested. Thanks for turning the previous round around so quickly — very close to done here.

[CryoThrust]

The previous fork run was green for unit tests, formatting, and all other jobs; only Doris connector integration test failed. I reran the failed job on the fork to distinguish an external integration flake from a regression. I will follow the rerun result before making code changes.

[CryoThrust]

The rerun completed successfully: the required Build check is now green, including the previously failing Doris connector integration job. The full CI matrix is passing. Please take another look when convenient.

[DanielLeens]

What Problem Does This PR Solve?

If the coordinator thread that is inserting a job into PeekBlockingQueue (the pending-job queue) gets interrupted — most concretely by CoordinatorService.clearCoordinatorService() calling executorService.shutdownNow() during a master step-down/failover — PeekBlockingQueue.put() used to silently swallow the InterruptedException, log one line, and return normally. CoordinatorService then proceeded as if the insert had succeeded: it advanced the physical plan to PENDING and, for submitJob, completed the client's submit future successfully, even though the job was never actually added to the queue. The job was effectively lost from the local scheduling view while everyone downstream believed the submission worked.

This PR (fixing #12010) makes PeekBlockingQueue.put() propagate InterruptedException (restoring the interrupt flag first) instead of swallowing it, and updates both call sites in CoordinatorService (submitJob and restoreJobFromMasterActiveSwitch) to stop advancing job state / reporting success when the insert didn't actually happen.

This is my third full pass on this PR. My first review was at df53e85b, my second at 515b0f73 (both COMMENTED, both stale now — a new commit landed on top). I re-read the entire diff against current dev from scratch rather than trusting my earlier notes, and independently re-verified the fork's own CI run for the current head rather than relying on the author's status comments.

1. Code Change Review

1.1 Core Logic Analysis

Files touched: PeekBlockingQueue.java, CoordinatorService.java, CoordinatorServiceTest.java, PeekBlockingQueueTest.java. I diffed the current head directly against dev (not against my previous review point) to make sure nothing besides what I already traced twice is in scope.

PeekBlockingQueue.put() (utils/PeekBlockingQueue.java:54-66): signature changed from void put(E) to void put(E) throws InterruptedException. On interrupt it does Thread.currentThread().interrupt(); throw e; instead of logging and swallowing — textbook-correct handling of an InterruptedException you don't want to fully absorb.

Call sites, re-verified line-for-line against dev:

• submitJob's pendingJobQueue.put(pendingJobInfo) call needed no code change — it was already wrapped in a broad try { ... } catch (Throwable e) { jobSubmitFuture.completeExceptionally(...); } on dev, with cleanup of runningJobInfoIMap/runningJobMasterMap/pendingJobQueue already gated behind !jobSubmitFuture.isCompletedExceptionally(). The only thing missing was put() actually throwing, which this PR supplies.
• CoordinatorService.java:1157-1165 (restoreJobFromMasterActiveSwitch) is the one call site with a real code change: now wrapped in an explicit catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new SeaTunnelEngineException(String.format("Job id %s restore interrupted while entering pending queue", jobId), e); }. updateJobState(JobStatus.PENDING) is correctly skipped because the method throws before reaching that line.

Re-grepped PeekBlockingQueue usage across the repo — still fully contained inside seatunnel-engine-server, so the checked-exception signature change has no ripple effect outside this module.

Diff since my last review (515b0f73 → 80d391f): I diffed these two commits directly and the only change is reformatting two lines in CoordinatorServiceTest.java (the long jobId = ... declaration and the ReflectionUtils.setField(...) call inside testInterruptedPendingJobInsertionDuringRestoreFailsRestore) from the manually-wrapped, multi-line shape my previous review flagged as a Spotless violation back into a single line each. No logic changed in this final commit — it is exactly the one-line/mechanical fix I asked for.

1.2 Compatibility Impact

Fully compatible.

• PeekBlockingQueue is an internal seatunnel-engine-server utility with no consumers outside that module.
• No config Option, public API, SPI, checkpoint/savepoint format, or serialization protocol is touched.
• The externally-visible behavior change is that a submission or restore that races with a master step-down and previously reported false success now correctly reports failure. Since the old behavior was itself the bug, this is a correctness fix, not a backward-incompatible change needing a migration note in docs/en/introduction/concepts/incompatible-changes.md.

1.3 Performance / Side-Effect Analysis

No new locks, no new polling loops, no additional allocations on the hot path beyond the try/catch itself. Thread.currentThread().interrupt() is idempotent if invoked twice.

Carried over from my earlier passes and re-verified once more: on the interrupted-put path, the JobMaster already constructed and init()-ed just before the failed put() is discarded without an explicit interrupt sweep reaching it (it isn't yet in runningJobMasterMap/pendingJobQueue at that point). JobMaster.init() only constructs a CheckpointManager without starting scheduled threads/timers at that stage, so this remains not a live-thread/resource leak — just normal garbage collection once the reference goes out of scope. Not a blocker.

1.4 Error Handling and Logging

• No swallowed exceptions remain in PeekBlockingQueue.put(); the log statement, @Slf4j, and the now-unused ExceptionUtils import were correctly removed together.
• No sensitive data is logged.
• Minor asymmetry, still present, non-blocking: submitJob's interrupted-put case still relies on the generic catch (Throwable e) instead of an explicit, jobId-scoped message like the restore path has. Both are diagnosable (a full stack trace is still logged); this is a log-grep-ability nicety only.

2. Code Quality Assessment

2.1 Coding Standards

Clean, minimal diff. Dead-import removal in PeekBlockingQueue.java is good hygiene. The formatting regression from my previous round is gone — I confirmed the two flagged lines are each within the 100-column limit now (checked with a direct character count, not just eyeballing), and this matches the fork's own Code style job passing (see CI verification below).

2.2 Test Coverage and Test Stability

Sorry for the back-and-forth — this is now clean. Recapping what's carried over and confirming current status directly against the current head rather than re-trusting my prior notes:

1. Round 1 blocker (PeekBlockingQueueTest.testClear() missing throws InterruptedException, breaking seatunnel-engine-server test-compile): still fixed as of this head — PeekBlockingQueueTest.java:118 declares throws InterruptedException, matching the other three methods in the file.
2. Round 2 blocker (self-inflicted Spotless formatting regression in the last commit): fixed exactly as described above — confirmed by direct diff against 515b0f73 and by CI (below).

Both regression tests are present and I re-traced their mechanics fresh rather than reusing my earlier read:

• testInterruptedPendingJobInsertionFailsSubmission (CoordinatorServiceTest.java:1339+, via InterruptiblePendingJobQueue): drives the interrupt through the real clearCoordinatorService() → executorService.shutdownNow() path while a real submit task is genuinely parked on releasePut.await() (a CountDownLatch, not a sleep), then calls through to the real PeekBlockingQueue.put(). Asserts the future completes exceptionally, the job never lands in the queue, and job state never advances to PENDING. Stable — no timing assumptions, no Thread.sleep.
• testInterruptedPendingJobInsertionDuringRestoreFailsRestore (CoordinatorServiceTest.java:1851+, via AlwaysInterruptedPendingJobQueue): deterministically throws InterruptedException from put() every time (no timing dependency at all), invokes the private restoreJobFromMasterActiveSwitch via the existing reflection helper, and asserts the reflection call surfaces InvocationTargetException → SeaTunnelEngineException → InterruptedException, that the job never enters the pending queue, and that job state never advances to PENDING. The finally { Thread.interrupted(); instance.shutdown(); } correctly clears the test thread's interrupt flag before shutdown, so it can't bleed into other tests sharing the thread. Stable.

Both call sites named in the PR description now have dedicated, deterministic regression coverage. I don't see any further coverage gap worth blocking on.

Independent CI verification (not just trusting the author's "CI is green" comments): I dereferenced the fork's own Actions run for the exact current head (80d391ff98cede63f9f6c732a09ba2fa4ac8bea0) — CryoThrust/seatunnel run 33749388227, conclusion success. I pulled the full job list rather than just the top-level status: Code style = success, License header = success, both unit-test JDK 8/11 on ubuntu = success, both unit-test JDK 8/11 on windows = success, and every non-skipped integration-test job (including the Doris job the author mentioned had flaked and been rerun) = success. This is a genuinely green build on the real fork run for this head, not the apache-side pointer check.

2.3 Documentation Updates

Not required. Purely internal engine-server correctness fix — no new/changed config Option, CLI flag, or public-facing behavior contract.

3. Architectural Soundness

3.1 Elegance of the Solution

"Propagate instead of swallow" remains the simplest correct fix and matches contract option 1 from issue #12010 (fail the interrupted/stale submission and preserve the interrupt flag). Unchanged note from my earlier passes: the issue text says the author would submit a PR "after maintainers confirm the intended interruption contract" — I don't see a recorded maintainer confirmation in the issue thread. That's a project-process question for the maintainers, not a code defect; option 1 is a safe, minimal, correct choice on its own merits.

3.2 Maintainability

Good — the interrupted-insert case is now an explicit, visible branch at both call sites instead of a silent no-op buried in a utility class.

3.3 Extensibility

No concerns; no new abstractions introduced.

3.4 Historical-Version Compatibility

No serialization, checkpoint/savepoint, or config format is touched. No impact on upgrade paths from previously released versions.

4. Issue Summary

#	Issue	Location	Severity	Status
1	testClear() missing throws InterruptedException, broke seatunnel-engine-server test-compile	PeekBlockingQueueTest.java:117-125	High	Fixed (carryover from round 1, confirmed resolved)
2	Self-inflicted Spotless formatting regression in the last commit of round 2, blocking Code style CI and gating unit-test	CoordinatorServiceTest.java	High	Fixed (carryover from round 2, confirmed resolved by direct diff + CI)
3	Minor asymmetry: submitJob's interrupted-put case uses the generic catch (Throwable e) instead of an explicit, jobId-scoped message like the restore path has	CoordinatorService.java (submit-path catch block)	Low	Open — optional polish, never blocking

No new issues were introduced by the latest (formatting-only) commit.

5. Merge Recommendation

Conclusion: Ready to merge

1. Blockers — must be fixed

   • None remaining. Both prior blockers (test-compile break, Spotless regression) are fixed and independently confirmed via the fork's own CI run for this exact head, not just the author's status comments.

2. Recommended fixes — non-blocking

   • Issue #3: optionally give submitJob the same explicit, jobId-scoped InterruptedException message the restore path already has, for log consistency. Purely cosmetic, not required for this PR to land.

Overall assessment: this has been a genuinely productive back-and-forth — every substantive gap I raised across two prior rounds (the pre-existing test-compile break, the missing restore-path regression test, and the self-inflicted formatting regression) has been closed, and I re-traced the fix and both new tests end-to-end on this pass rather than assuming my earlier notes still applied. The underlying fix itself has been correct and minimal since round 1: propagate instead of swallow, restore the interrupt flag, don't advance job state or report false success on failure. Both regression tests are deterministic and drive the real production interrupt path rather than taking shortcuts. Thank you for turning around three rounds of feedback quickly and precisely — this is a solid first contribution, and I'm glad to see it land once a maintainer takes a final look.