[Fix][Zeta] Reuse fixed slots after master failover

[zhangshenghang]

What does this PR do?

Fixes #11437. During active-master failover, a restored JobMaster now reuses the matching active worker slots recorded for the job instead of requesting them again. This lets fixed-slot jobs recover while workers still retain their task groups.

Why is this needed?

The replacement master previously treated retained slots as unavailable and left the restored job pending. If only some slots can be reused, cleanup releases only newly allocated slots and preserves the active retained slots for the next retry.

Tests

• Added a fixed-slot master-failover regression test that exhausts all free slots before restore.
• ./mvnw -pl seatunnel-engine/seatunnel-engine-server -DskipITs -Dtest=JobMasterMasterFailoverResourceTest test
• ./mvnw -q -DskipTests verify

[nielifeng]

I validated this PR commit (46d2f3909f7e97f8f516c8fba68d12df471d7718) against the fixed-slot reproduction from #11437.

Environment: 3 Masters, 2 Workers, 16 fixed slots, WAIT scheduling. Workload: FakeSource -> Console, parallelism 10, 20,000 rows/split, 11 assigned / 5 free slots. Package SHA-256: 9bd831336fc547aed1b53adda4af636a3ec1771cfdab4c81b9689c70e7d73c91.

Three consecutive Active Master Java-process failures passed:

• election: 1 s, 1 s, 2 s
• job visible as RUNNING: 5 s in every round
• source and sink QPS >= 90% of the 87.9k rows/s pre-fault median: 5 s in every round
• slots: exactly 11 assigned / 5 free in every post-fault sample
• checkpoints: 2 before fault injection -> 19 completed after round 3, 0 failed

After all Masters rejoined, cancelling the job returned all resources in 2 seconds: 16/16 slots free, no running/pending jobs, and empty Worker runningJobIds. Neither Worker was restarted.

I found no RESOURCE_NOT_ENOUGH, NoEnoughResourceException, duplicate TaskGroupLocation, or stale-slot message. Only transient Hazelcast member-left/connection errors occurred at the injected failures.

For this topology and workload, the PR fixes both the failover restore and cancellation cleanup symptoms reported in #11437.

[SEZ9]

Thanks for working on this. I re-reviewed the latest head 46d2f3909f7e from scratch.

What this PR fixes

• User pain: After an active-master failover, a fixed-slot job could never recover — the replacement master re-requested slots that workers still retained for the job, saw them as unavailable, and left the restored job stuck pending (#11437).
• Fix approach: JobMaster.init(..., restart=true) now sets a masterFailoverRestore flag; during preApplyResources, each task/coordinator first checks getReusableSlot() (owned slot profile from the IMap + resourceManager.slotActiveCheck) and completes the future with the retained slot instead of applying anew, while the failure/cleanup path filters reusedSlotProfiles out of releaseResources so retained slots survive a partial-reuse retry.
• One-line summary: The core reuse logic is sound and well-tested for the happy path, but the flag is cleared too eagerly and isn't volatile, the release-path filtering leans on SlotProfile.equals in an O(n·m) contains, and the new test has resource/cleanup hygiene issues — all fixable before merge.

Runtime chain I rechecked

CoordinatorService restore path (restart=true)
  -> JobMaster.init(initializationTimestamp, restart=true)   JobMaster.java:230  (sets masterFailoverRestore = restart, JobMaster.java:232)
  -> JobMaster.preApplyResources(subPlan)                     JobMaster.java:519
     -> preApplyResourcesForAll(futures, reusedSlotProfiles)  JobMaster.java:642  (whole-job path, iterates physicalPlan.getPipelineList())
        -> preApplyResourcesForSubPlan(subPlan, futures, reusedSlotProfiles)  JobMaster.java:651
           -> getReusableSlot(taskGroupLocation)              JobMaster.java:701
              -> getOwnedSlotProfiles(taskGroupLocation)      (reads ownedSlotProfilesIMap persisted before deployment)
              -> resourceManager.slotActiveCheck(slotProfile) (confirms worker still holds the slot sequence)
           -> ResourceUtils.applyResourceForTask(...) only when no reusable slot; otherwise CompletableFuture.completedFuture(reusableSlot)
     -> success branch: physicalPlan.setPreApplyResourceFutures(futures); masterFailoverRestore = false  JobMaster.java:586-587
     -> failure branch: resourceManager.releaseResources(jobId, joined futures .filter(p -> !reusedSlotProfiles.contains(p)))  JobMaster.java:616-623  (skips releasing retained worker slots so the next retry can still reuse them)

Findings

Issue 1: Release-path filtering relies on List.contains / SlotProfile.equals, which is fragile and O(n·m)

• Location: seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/master/JobMaster.java:621
• Why it matters: - Root cause: in the failure branch, newly applied slots are released via .filter(slotProfile -> !reusedSlotProfiles.contains(slotProfile)) on a List<SlotProfile>. Correctness depends entirely on SlotProfile implementing value-based equals/hashCode; if it does not (or if two profiles for the same worker/sequence compare unequal after deserialization), a reused retained slot would be released here, killing the still-running task groups on the worker instead of preserving them for the next retry.
• Impact: releasing a retained slot during a partial-failure cleanup would break the very recovery path this PR introduces, and with large parallelism the linear contains on a list adds avoidable O(n·m) cost.
• Fix: since reused futures are always CompletableFuture.completedFuture(reusableSlot) holding the exact same object, collect them into a Set (e.g. Collections.newSetFromMap(new IdentityHashMap<>()) or a HashSet after confirming SlotProfile has a proper equals) and filter against that set. Please also add a test asserting that a partial failure only releases newly allocated slots.
• Evidence: seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/master/JobMaster.java:620 slotProfile ->; seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/master/JobMaster.java:621 !reusedSlotProfiles.contains(; seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/master/JobMaster.java:526 List<SlotProfile> reusedSlotProfiles = new ArrayList<>();
• Suggested fix: This filter depends on SlotProfile.equals behaving as value equality; if it doesn't, a reused retained slot would be released in the partial-failure path, which defeats the fix. Since the reused futures hold the exact object you put into reusedSlotProfiles, an identity Set (Collections.newSetFromMap(new IdentityHashMap<>())) would be both safer and faster than List.contains at high parallelism.
• Severity: Medium

Issue 2: New masterFailoverRestore flag is not volatile although it is written and read across threads

• Location: seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/master/JobMaster.java:186
• Why it matters: - Root cause: the new field masterFailoverRestore is a plain boolean, but it is written in the synchronized init(...) and later mutated in preApplyResources(...) (masterFailoverRestore = false;) and read in getReusableSlot(...). preApplyResources is invoked from scheduler/executor threads, not necessarily the thread that ran init, so there is no happens-before guarantee for the write in preApplyResources.
• Impact: on Zeta, a stale read could either cause the restored master to skip slot reuse (job stays pending, reproducing #11437) or keep reusing after the flag was cleared, releasing/holding fixed slots inconsistently during retries.
• Fix: declare the field private volatile boolean masterFailoverRestore;, matching the existing convention right above it where needRestore is declared @Getter private volatile boolean needRestore = true;.
• Evidence: seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/master/JobMaster.java:232 this.masterFailoverRestore = restart;; seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/master/JobMaster.java:587 masterFailoverRestore = false;
• Suggested fix: This field is written in init(...) (synchronized) but also mutated in preApplyResources(...) and read in getReusableSlot(...), which run on scheduler threads. Please make it volatile, consistent with needRestore just above (@Getter private volatile boolean needRestore = true;), to avoid stale reads deciding whether retained slots are reused.
• Severity: Medium

Issue 3: masterFailoverRestore is cleared after the first successful preApply, so later subPlan-level restores stop reusing retained slots

• Location: seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/master/JobMaster.java:587
• Why it matters: - Root cause: masterFailoverRestore = false; sits in the success branch of preApplyResources(SubPlan). That method can be called per-pipeline (isSubPlan == true), so after the first pipeline of a multi-pipeline job successfully pre-applies, the flag flips for the whole JobMaster.
• Impact: for a restored job with multiple pipelines that are scheduled/pre-applied independently, only the first preApplyResources call benefits from slot reuse; remaining pipelines fall back to requesting fresh fixed slots that the workers still own, which is exactly the deadlock #11437 describes.
• Fix: only clear the flag when the whole-job path succeeded (!isSubPlan), or track reuse eligibility per PipelineLocation instead of a single job-level boolean, so each pipeline gets one reuse opportunity after failover.
• Evidence: seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/master/JobMaster.java:530 preApplyResourcesForSubPlan(subPlan, preApplyResourceFutures, reusedSlotProfiles);
• Suggested fix: Clearing masterFailoverRestore here affects the whole JobMaster, but preApplyResources(subPlan) can be invoked per pipeline. After the first pipeline succeeds, later pipelines of the same restored job would no longer reuse their retained fixed slots and could hit the same deadlock as #11437. Consider clearing the flag only on the whole-job path (!isSubPlan) or tracking reuse per PipelineLocation.
• Severity: Medium

Issue 4: Test depends on a config resource not added in this PR and leaves entries in shared IMaps

• Location: seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/master/JobMasterMasterFailoverResourceTest.java:99
• Why it matters: - Root cause: newJobMaster references stream_fake_to_inmemory_with_sleep.conf via TestUtils.createTestLogicalPlan, but this PR only touches two files, so the test silently depends on that resource already existing in seatunnel-engine-server test resources; if it was only present in another module, the test fails at init.
• Additionally, persistPreAppliedSlots writes into the Hazelcast maps ownedSlotProfilesIMap, runningJobState, etc., and while releasePersistedSlots releases the slots, the IMap entries themselves are never removed, which can leak state into other tests sharing the same AbstractSeaTunnelServerTest instance (the same class of flakiness we fixed in PR #11425).
• Fix: confirm the conf file exists under this module's src/test/resources, and clear the populated ownedSlotProfilesIMap entries (e.g. map.clear() or remove the job's pipeline keys) in the finally block.
• Evidence: seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/master/JobMasterMasterFailoverResourceTest.java:88 resourceManager.releaseResources(jobId + 1, blockerSlots).join();; seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/master/JobMasterMasterFailoverResourceTest.java:135 jobMaster.setOwnedSlotProfiles(subPlan.getPipelineLocation(), slotProfiles);
• Suggested fix: Two small things: (1) stream_fake_to_inmemory_with_sleep.conf isn't added in this PR — please make sure it exists in this module's src/test/resources, otherwise the test fails at DAG creation. (2) persistPreAppliedSlots populates ownedSlotProfilesIMap but the finally block only releases slots; please also remove those map entries so state doesn't leak into other tests on the shared server instance.
• Severity: Low

Issue 5: New reusedSlotProfiles out-parameter on the private pre-apply helpers is undocumented

• Location: seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/master/JobMaster.java:654
• Why it matters: - Root cause: preApplyResourcesForAll and preApplyResourcesForSubPlan gained a third parameter List<SlotProfile> reusedSlotProfiles that is mutated as an output collection, but neither method documents this contract, while the nearby new getReusableSlot got a full Javadoc.
• Impact: future maintainers of JobMaster (already a large class) can easily miss that the caller relies on this list being populated to protect reused slots in the release path; an accidental refactor that stops appending to it would silently release retained slots.
• Fix: add a short Javadoc @param reusedSlotProfiles collector for slots reused from the previous master's owned-slot mapping; these must never be released on partial failure, or refactor the helpers to return the reused slots instead of mutating a caller-owned list.
• Evidence: seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/master/JobMaster.java:644 List<SlotProfile> reusedSlotProfiles) {; seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/master/JobMaster.java:654 List<SlotProfile> reusedSlotProfiles) {
• Suggested fix: Nice Javadoc on getReusableSlot below — could you add a matching @param reusedSlotProfiles note here (and on preApplyResourcesForAll)? It's an out-parameter the caller depends on to avoid releasing reused slots on partial failure, and that contract isn't obvious from the signature.
• Severity: Low

Issue 6: Behavior change in master-failover recovery is not reflected in Zeta fault-tolerance docs or release notes

• Location: seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/master/JobMaster.java:697 private SlotProfile getReusableSlot(TaskGroupLocation taskGroupLocation) {
• Why it matters: - Root cause: this PR changes observable engine behavior — after an active-master switch, restored jobs with fixed slots now reuse retained worker slots instead of hanging pending — but no documentation or release-note entry accompanies the code change.
• Impact: operators debugging failover with dynamic-slot: false won't find the new recovery semantics (including the partial-reuse cleanup behavior described in the PR body) documented anywhere, and the fix won't surface in the next release notes.
• Fix: add a line to the release-note/changelog for the next release referencing #11437, and consider a short paragraph in the Zeta deployment/fault-tolerance docs describing that fixed slots retained by workers are reused after master failover once slotActiveCheck confirms them.
• Evidence: seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/master/JobMaster.java:702 return slotProfile != null && resourceManager.slotActiveCheck(slotProfile)
• Severity: Low

Review conclusion

Conclusion: can merge; the notes below are non-blocking

2. Suggested (non-blocking) follow-ups

• Issue 1: Release-path filtering relies on List.contains / SlotProfile.equals, which is fragile and O(n·m)
• Issue 2: New masterFailoverRestore flag is not volatile although it is written and read across threads
• Issue 3: masterFailoverRestore is cleared after the first successful preApply, so later subPlan-level restores stop reusing retained slots
• Issue 4: Test depends on a config resource not added in this PR and leaves entries in shared IMaps
• Issue 5: New reusedSlotProfiles out-parameter on the private pre-apply helpers is undocumented
• Issue 6: Behavior change in master-failover recovery is not reflected in Zeta fault-tolerance docs or release notes

Nice work — thanks again for the contribution. Happy to discuss any of the above.

[nielifeng]

Reviewed the diff and traced the call paths on dev. The approach looks solid to me:

• getReusableSlot doesn't blindly trust the persisted ownedSlotProfilesIMap entry — it confirms with resourceManager.slotActiveCheck(slotProfile) that the worker still holds the matching slot sequence before reusing. That's the right guard against a stale slot record after failover.
• The cleanup filter !reusedSlotProfiles.contains(slotProfile) correctly preserves reused slots and only releases newly-applied ones. I checked SlotProfile.equals/hashCode (slotID + worker + sequence) — identity is stable, so contains is reliable here.
• The regression test genuinely exercises the fix: it exhausts all free fixed slots (assertEquals(0, getUnassignedSlots)) before restore, so the old code would leave the job pending and only the reuse path lets it recover. dynamicSlot=false + fixed slotNum targets the right scenario.

I also double-checked the multi-pipeline concern: master failover recovery goes through CoordinatorService#pendingJobSchedule → preApplyResources() (no-arg, all pipelines in one call), and masterFailoverRestore is only cleared after that whole call succeeds — so a JobMaster-level flag covers all pipelines correctly. The per-SubPlan preApplyResources(this) path is normal-operation pipeline restart where reuse is intentionally skipped. So no issue there.

One minor suggestion (non-blocking): masterFailoverRestore is a plain boolean. It's written in init() and in preApplyResources, and read from the per-SubPlan path which runs on executorService threads. For consistency with the adjacent volatile boolean needRestore, it might be worth marking it volatile to avoid any visibility gap. Low priority given the coordinator-thread write happens well before pipeline restarts.

Nice fix overall.

[DanielLeens]

Thanks for the update. I rechecked the latest head 46d2f3909f7e from scratch around the actual restore path (SubPlan -> JobMaster.preApplyResources(this) -> getReusableSlot(...) -> ResourceUtils.applyResourceForPipeline(...)), and I do not see a new source-level blocker from Daniel's side on the current diff.

I also rechecked @SEZ9's earlier notes against the current code:

• SlotProfile already has value-based equals/hashCode in seatunnel-engine-server/.../SlotProfile.java:83-99, so the release-path filter is not missing equality support.
• masterFailoverRestore is only cleared on the whole-job success branch in JobMaster.java:580-587, not on the per-subplan restore branch, so the current head does not drop reuse after the first restored pipeline.

The remaining gates I see right now are operational rather than source-level:

• the PR is still in draft state
• the branch is behind dev (ahead_by=1, behind_by=3, compare status diverged)
• the current failing CI jobs are all-connectors-it-7 (8, ubuntu-latest) and kafka-connector-it (8, ubuntu-latest) on the fork run, which do not look tied to this JobMaster-only diff yet

Given that, the lowest-cost next step is to sync with the latest dev, rerun CI, and then flip the PR out of draft once the updated checks settle. If those same lanes still fail on the refreshed head, please drop the new run link here and I can help narrow them down further.