fix(RayAppMaster): refactor executor state in RayAppMaster.

[my-vegetable-has-exploded]

What does this PR do?

The old code counted Spark executor generations instead of logical Ray actors. A restarted
executor was recorded in both executors and restartedExecutors, while stale restart mappings
survived executor removal. This allowed registeredExecutors to report a deficit while the
restart-based limit check simultaneously reported that the application was full.
RequestExecutors would then acknowledge the target without creating any actor. Tracking slots
by the stable Ray actorId removes this contradiction: a restart changes the Spark executor
generation, not the number of logical actors.

Two lifecycle details are essential to make that model correct. First, the actor-slot registry must
retain an ActorHandle across executor disconnects; otherwise Ray's reference can be released and
the non-detached actor terminated while RayDP still counts its slot. Second, RayAppMaster must
reconcile after KillExecutors: Spark treats the requested total as persistent and expects the
cluster manager, as YARN does, to replace missing executors without another identical request from
the driver.

Why is this change needed?

Problem

RayDP previously conflated two different identities:

• A Ray actor has a stable identity and keeps the same named actorId across restarts.
• A Spark executor gets a new executorId for every restarted generation.

The previous implementation used several generation-based states to approximate the number of
executor actors:

• registeredExecutors to calculate how many executors were missing.
• executors.size + restartedExecutors.size to enforce the executor limit.
• restartedExecutors to retain the mapping from a new executor ID to its original actor ID.

These states have different lifecycles and can disagree. In particular, a restarted executor is
present in both executors and restartedExecutors, so the same Ray actor is counted twice.
Moreover, entries in restartedExecutors are not removed when the corresponding executor
generation disconnects, allowing historical restarts to remain in executor accounting.

Failure Scenario

The incorrect count follows directly from three asymmetric code paths:

1. When a restarted actor requests a new Spark executor ID, RayAppMaster calls
   addPendingRegisterExecutor(newExecutorId, ...), which inserts the new generation into
   appInfo.executors. It then also writes
   restartedExecutors(newExecutorId) = originalActorId.
2. Every newExecutorId is unique. A later restart therefore adds another map entry instead of
   replacing the entry for the previous generation, even though both values point to the same
   original Ray actor.
3. On RPC disconnect, ApplicationInfo.kill() removes the current generation from
   appInfo.executors. It cannot remove the matching restartedExecutors entry because that map
   is owned separately by RayAppMaster, and neither the disconnect path nor KillExecutors
   performs such cleanup.

The capacity check nevertheless uses:

appInfo.executors.size + restartedExecutors.size

This is wrong in two different phases. While a restarted generation is active, its executor ID is
present in both maps and is counted twice. After it disconnects, it is removed from executors but
remains in restartedExecutors, so it becomes a permanently counted historical generation.

For one logical actor with original actorId = 7, the state evolves as follows:

Event	appInfo.executors	restartedExecutors	Count used by limit check	Logical actor slots
Initial generation	{7}	{}	1	1
Restart as executor 1001	{1001}	{1001 -> 7}	2	1
Executor 1001 disconnects	{}	{1001 -> 7}	1	1
Restart as executor 1002	{1002}	{1001 -> 7, 1002 -> 7}	3	1

Thus, each restart adds another capacity unit to restartedExecutors, although the number of Ray
actor slots never changes. Disconnecting the generation does not undo that increase.

The restart-registration path does not call requestNewExecutor(), so it bypasses the maximum
check that later consumes this inflated count. Each disconnect also decrements
registeredExecutors, making remainingUnRegisteredExecutors > 0 true and admitting the next
restart generation. The history can therefore grow to the configured maximum and beyond.

The key contradiction is:

registeredExecutors says executors are missing, while restartedExecutors says the
application is already at its limit.

Both cannot be valid measures of Ray executor capacity.

Solution

Track logical Ray actor slots independently from Spark executor generations.

An actor slot is keyed by the original actorId:

• Creating a new Ray actor adds one actor slot.
• Restarting that actor creates a new Spark executorId, but does not add another slot.
• An executor disconnect removes only that Spark executor generation because Ray may restart the
  same actor.
• When Spark sends KillExecutors, RayDP removes the actor slot and requests actor shutdown.
• The Spark driver's requested total is stored as desiredExecutors.

These rules describe two stages that may occur in the same failure path. The AppMaster's initial
RPC disconnect handling keeps the slot because Ray may recover the actor. If Spark subsequently
decides to terminate that executor and sends KillExecutors, RayDP then removes the slot and
stops the actor. Therefore slot removal is tied to the KillExecutors command, not only to an
explicit scale-down reason.

Executor reconciliation then becomes:

actorsToAdd = max(0, desiredExecutors - actorSlots)

This compares two values with matching semantics: the number of logical actors requested by Spark
and the number of logical Ray actor slots currently owned by the application.

The implementation keeps an actorId -> ActorHandle registry as the actor-slot state and maps each
Spark executor generation back to its stable actor ID. The restarted-executor mapping required by
recoverable RDD lookup is derived from this state, but is no longer used for capacity accounting.

The actor-slot registry cannot continue to use the previous executorIdToHandler map. When an
executor disconnects, onDisconnected calls ApplicationInfo.kill(..., shutdownActor = false) to
remove that Spark executor generation while allowing Ray to restart its actor. However, kill()
unconditionally executes executorIdToHandler -= executorId. Removing the executor ID also removes
the map entry that retains the corresponding actor handle. If this is the last retained handle, the
handle becomes unreachable and the actor can be terminated, which violates the expected restart
semantics of shutdownActor = false.

The handle is not only a lookup convenience: it keeps Ray's reference to a non-detached actor
alive. RayDP creates a named actor with setName, but does not give it ActorLifetime.DETACHED.
In Ray 2.1,
NativeActorHandleReference.finalizeReferent() calls nativeRemoveActorHandleReference() after a
Java handle becomes unreachable. Losing the last retained handle can therefore reduce the
distributed reference count to zero and terminate the actor. setMaxRestarts(-1) allows recovery
from actor process failures; it does not make the actor independent of its owning reference.

The sharp failure sequence is:

actorId = 23, current Spark executorId = 1001
executorIdToHandler(1001) = handle(23)

executor 1001 disconnects
  -> kill(1001, shutdownActor = false)
  -> actorSlots still contains 23, because Ray is expected to restart it
  -> executorIdToHandler removes 1001 unconditionally
  -> handle(23) becomes unreachable and is garbage-collected
  -> Ray removes the actor-handle reference and actor 23 can be terminated

RayDP is then left with a ghost slot: actorSlots says actor 23 exists, while the actor and its
handle are gone. Because reconciliation sees the slot as present, it creates no replacement. A
named actor is not sufficient protection here; naming makes lookup possible only while the actor
is still alive.

For this reason, f61f80e5e6e0daa125172573ecf4c8262aea2af1 replaces
executorIdToHandler with two maps:

executorId -> actorId
actorId    -> ActorHandle

The first map follows the lifetime of a Spark executor generation. The second is both the actor-slot
registry and the strong-reference holder for the Ray actor. Removing a disconnected executor ID
now deletes only the generation mapping. The stable actor handle remains until the actor slot itself
is deliberately removed, so Ray can restart the actor and register a new Spark executorId.

Reconcile Immediately after KillExecutors

Spark's total-executor request is a persistent desired-state contract, not a one-shot request to
create a delta. CoarseGrainedSchedulerBackend explicitly assumes that the cluster manager will
eventually fulfill an acknowledged total request when resources become available. Consequently,
the Spark driver does not periodically resend the same target merely because an executor
disappears; the resource-management side is responsible for retaining the target and reconciling
its actual resources against it.

YARN demonstrates this contract directly. YarnAllocator stores the driver's total in
targetNumExecutorsPerResourceProfileId. On every allocator heartbeat, it computes:

missing = target - pending - running - starting

When a container exits, the running count decreases and the next heartbeat requests a replacement,
without requiring the Spark driver to resend its unchanged target.

Before this commit, RayDP retained desiredExecutors, but reconciled only when it received
RequestExecutors. Consider a target of 1000 and Spark calling killAndReplaceExecutor(17). Spark
sends KillExecutors with adjustTargetNumExecutors = false, so its desired target correctly stays
at 1000. RayAppMaster removes actor slot 17 and acknowledges the kill, leaving 999 slots. No new
RequestExecutors(1000) is required by Spark's contract, and RayAppMaster had neither a periodic
allocator loop nor a reconciliation call on this state transition. The application could therefore
remain at 999 indefinitely; repeated replacement kills could accumulate the deficit.

This commit invokes reconcileExecutors() immediately after processing KillExecutors:

• For a real scale-down, Spark first lowers the total request, so desiredExecutors already matches
  the reduced slot count and no replacement is created.
• For kill-and-replace or health-related removal, Spark keeps the total request unchanged, so the
  removed slot immediately appears as a deficit and RayDP creates a replacement.

RequestExecutors and KillExecutors are the two events that change RayAppMaster's target or actor
slot count. Reconciling on both events gives RayDP the same desired-state behavior that Spark expects
from resource managers such as YARN, without requiring a periodic target synchronization from the
driver.

Related Issue

Fixes #

Type of Change

• Bug fix
• New feature
• Documentation update
• Test update
• Refactoring
• Build / CI / dependency change
• Other

How was this tested?

This fix has been tested in our production environment and works well with dynamic execution enabled and 1,000 executors. Before this fix, large-scale jobs would occasionally fail.

Checklist

• I have added or updated tests if needed
• I have updated documentation if needed
• I have run relevant tests locally
• I have checked that this change is backward compatible

[my-vegetable-has-exploded]

Hi @pang-wu @carsonwang , could you please review this PR? This fix addresses a real issue we encountered in production.

[my-vegetable-has-exploded]

@claude review