[Bug][Zeta] CoordinatorService job-scheduling thread can wedge permanently when a REJECT-failed job's cleanup hangs

### What happened

Zeta's job-scheduling completion path can permanently hang the single thread (`pending-job-schedule-runner`) that services **all** job scheduling for a master node. When it hangs, the triggering job never resolves past `JobStatus.UNKNOWABLE` (it never reaches a stable `FAILED` with a history record), and — because scheduling for every subsequent job submission on that master also runs through this same thread — the cluster's ability to schedule new jobs may be silently frozen from that point on, with no exception thrown and no error logged.

This was found while building an E2E regression test for an unrelated scenario (multi-pipeline restore contention after a worker is killed); the reproduction below is a side effect of that test's own setup phase, before its actual scenario ever runs.

### Root cause (traced from source + real CI logs, not yet confirmed with a live thread dump — see "what's still needed" below)

`CoordinatorService.completeFailJob()` (`seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/CoordinatorService.java:510-525`) blocks on `jobMaster.getJobMasterCompleteFuture().join()` (line 520). The entire completion chain feeding that future runs **synchronously on the calling thread** — there is no executor hop anywhere in it:

- `PhysicalPlan.completeJobEndFuture()` (`PhysicalPlan.java:441-443`) calls `jobEndFuture.complete(...)` directly.
- `org.apache.seatunnel.engine.common.utils.concurrent.CompletableFuture.whenComplete()` (`CompletableFuture.java:102-104`) and `PassiveCompletableFuture`'s constructor (`PassiveCompletableFuture.java:30-41`) both use plain (non-`Async`) `whenComplete`, so the listener chain fires synchronously.
- This synchronously drives `JobMaster.initStateFuture()`'s completion callback (`JobMaster.java:488-497`), which calls `cleanJob()` (`JobMaster.java:966-971`) before completing `jobMasterCompleteFuture`.

`cleanJob()`'s steps include `checkpointManager.clearCheckpointIfNeed()`, `getJobDAGInfo()` (`JobMaster.java:856-877`, guarded by `synchronized(this)` — the **same monitor** used by `cancelJob()`/`stopJob()`), `jobHistoryService.storeJobInfo()`/`storeFinishedJobState()` (the actual `finishedJobStateImap.put()`, `JobHistoryService.java:257-261`), and `scheduleRemoveJobStateMaps()` (`JobMaster.java:825-838`, a synchronous `pendingJobCleanupIMap.put()`).

**If any single step inside `cleanJob()` blocks instead of returning, `jobMasterCompleteFuture` never completes, `completeFailJob()`'s `.join()` blocks forever, and the calling thread — `pending-job-schedule-runner`, the only thread that runs `CoordinatorService.pendingJobSchedule()` — is wedged permanently.**

This is a `REJECT`-schedule-strategy path specifically: a job whose `preApplyResources()` gets only a partial slot grant fails immediately and terminally under `REJECT` (`CoordinatorService.java:463` → `:510`), with no retry (contrast `WAIT`, which sleeps and requeues). This is the engine's config-file default strategy.

### Evidence (fork `davidzollo/seatunnel`, Actions run `34181138685`, job `101921574204`, JDK 11)

Job id `1149543734560423937` needed 12 of 16 total fixed slots across 2 workers; `preApplyResources()` got a partial grant (5 of 12 succeeded). Its complete log footprint (18 lines total) ends at `03:16:23.9357504Z`, immediately after the second `updateJobState(FAILED)` call inside `completeFailJob()`. **`completeFailJob()`'s own final log statement — `"The job %s is not running because the resources is not enough insufficient"` (`CoordinatorService.java:521-524`), reachable only *after* `.join()` returns — never appears for this job anywhere in the (315,221-line) log.**

Three sibling jobs failed via the identical REJECT path in the same run (`1149552276294074369`, `1149552325602312193`, `1149552375816519681`) and **do** show that final log line — i.e. `.join()` returned normally for them. Something specific to job `1149543734560423937` (plausibly its unusually large 12-slot/multi-pipeline fan-out, or a timing overlap with a concurrent operation on the same `JobMaster`) caused its `cleanJob()` to hang where the siblings' did not.

Ruled out as causes: an escaped `Throwable` on the scheduler thread (`CoordinatorService.java:323`'s `"Error in pending job schedule thread"` never appears near the incident); Hazelcast's own callback-exception-swallowing wrapper (`ExceptionUtil.withTryCatch`'s `"Exception during callback"` string appears 14 times elsewhere in this same log — confirming the mechanism is live and working — but zero times for this job); a lost/reordered IMap write (the relevant maps are the same object references throughout, no evidence of desync); stale `PeekBlockingQueue` bookkeeping (read in full, no bug found).

### Blast radius

If confirmed, this is not just "one job's status becomes unqueryable." Because `cleanJob()` runs on the shared, single `pending-job-schedule-runner` thread, a hang here could silently freeze scheduling for **every subsequent job submitted to that master**, with no crash, no exception, and no error logged anywhere — an operator would only notice new jobs never starting, with no obvious cause in the logs.

### Two candidate blocking points (not yet distinguished)

1. `getJobDAGInfo()`'s `synchronized(this)` (`JobMaster.java:856-877`) contending indefinitely with a concurrent `cancelJob()`/`stopJob()` holding the same monitor on the same `JobMaster` instance.
2. A stalled Hazelcast IMap operation — either the `finishedJobStateImap.put()` in `storeFinishedJobState()`, or a per-vertex `runningJobStateIMap.get()` loop inside whatever builds the job's DAG/state summary — potentially exposed specifically by this job's large fan-out (12 slots / multiple pipelines), which would make this a scale-dependent trigger rather than a purely random one.

### What's still needed to close this out

A thread dump (`jstack`) of the master JVM captured while `pending-job-schedule-runner` is confirmed wedged, during a live reproduction. Static/log analysis alone cannot distinguish between the two candidates above. The E2E test that originally surfaced this (`SplitClusterFaultToleranceIT#testManyPipelinesRestoreContentionInWorkerDown`, from PR #12109 — a 4-pipeline, 12-slot job submitted against a 2-worker/16-slot fixed-pool cluster under the default `REJECT` strategy) is a plausible starting point for reproduction, though it did not hang on every run (3 sibling jobs completed the same failure path normally in the same CI run), suggesting the trigger condition is timing-sensitive rather than deterministic on every partial-grant failure.

### Fix containment (once the exact blocking call is known)

Expected to be narrow: a bounded timeout around `cleanJob()`'s blocking calls (or around the `.join()` in `completeFailJob()` itself, with a fallback status write on timeout), confined to `CoordinatorService.java` and/or `JobMaster.java`. No architectural change anticipated — but this should not be assumed without confirming the actual blocking call first, since a timeout alone would mask the symptom without fixing whatever is actually stalling (e.g. real lock contention that could still starve other work).

Start with CoordinatorService.completeFailJob(), JobMaster.cleanJob(), PhysicalPlan.completeJobEndFuture(), and the related CompletableFuture callbacks cited in the report. Run SplitClusterFaultToleranceIT#testManyPipelinesRestoreContentionInWorkerDown and capture a jstack while pending-job-schedule-runner is wedged. Done requires identifying which blocking call prevents jobMasterCompleteFuture from completing and confirming the contained fix with a regression test.

Written by the indexing model from the issue text.