From 49ee1bd014160ee1b85f9c18484b4f25b8ef759b Mon Sep 17 00:00:00 2001 From: epsilonwang Date: Fri, 16 Oct 2026 09:30:06 +0000 Subject: [PATCH 1/2] fix: retain executor owner handles across restarts Signed-off-by: epsilonwang --- .../spark/deploy/raydp/RayAppMaster.scala | 27 ++-- .../spark/deploy/raydp/RayAppMasterTest.scala | 142 ++++++++++++++++++ 2 files changed, 154 insertions(+), 15 deletions(-) create mode 100644 core/raydp-main/src/test/scala/org/apache/spark/deploy/raydp/RayAppMasterTest.scala diff --git a/core/raydp-main/src/main/scala/org/apache/spark/deploy/raydp/RayAppMaster.scala b/core/raydp-main/src/main/scala/org/apache/spark/deploy/raydp/RayAppMaster.scala index 6e578dc8..d05f5f67 100644 --- a/core/raydp-main/src/main/scala/org/apache/spark/deploy/raydp/RayAppMaster.scala +++ b/core/raydp-main/src/main/scala/org/apache/spark/deploy/raydp/RayAppMaster.scala @@ -199,21 +199,18 @@ class RayAppMaster(host: String, context.reply(success) case RequestAddPendingRestartedExecutor(actorId) => - val cores = appInfo.desc.coresPerExecutor.getOrElse(1) - val memory = appInfo.desc.memoryPerExecutorMB - // ray actor will restart using the old ID - val handlerOpt = Ray.getActor("raydp-executor-" + actorId) - if (!handlerOpt.isPresent) { - context.reply(AddPendingRestartedExecutorReply(None)) - } else if (!appInfo.actorSlots.contains(actorId)) { - // The actor may still be visible in Ray after Spark has scaled the slot down. - // Do not allow a late restart request to recreate a removed executor. - context.reply(AddPendingRestartedExecutorReply(None)) - } else { - val newExecutorId = s"${appInfo.getNextExecutorId()}" - val handler = handlerOpt.get.asInstanceOf[ActorHandle[RayDPExecutor]] - appInfo.addPendingRegisterExecutor(newExecutorId, actorId, handler, cores, memory) - context.reply(AddPendingRestartedExecutorReply(Some(newExecutorId))) + appInfo.actorSlots.get(actorId) match { + case Some(slot) => + // Keep the original owner handle reachable across executor generations instead of + // replacing it with a named-lookup handle when the same Ray actor restarts. + val cores = appInfo.desc.coresPerExecutor.getOrElse(1) + val memory = appInfo.desc.memoryPerExecutorMB + val newExecutorId = s"${appInfo.getNextExecutorId()}" + appInfo.addPendingRegisterExecutor(newExecutorId, actorId, slot.handle, cores, memory) + context.reply(AddPendingRestartedExecutorReply(Some(newExecutorId))) + case None => + // Do not allow a late restart request to recreate a removed actor slot. + context.reply(AddPendingRestartedExecutorReply(None)) } } diff --git a/core/raydp-main/src/test/scala/org/apache/spark/deploy/raydp/RayAppMasterTest.scala b/core/raydp-main/src/test/scala/org/apache/spark/deploy/raydp/RayAppMasterTest.scala new file mode 100644 index 00000000..644f7a05 --- /dev/null +++ b/core/raydp-main/src/test/scala/org/apache/spark/deploy/raydp/RayAppMasterTest.scala @@ -0,0 +1,142 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.deploy.raydp + +import java.lang.reflect.{InvocationHandler, Method, Proxy} +import java.util.Date + +import io.ray.api.ActorHandle +import org.junit.jupiter.api.Assertions.{assertEquals, assertSame, assertTrue} +import org.junit.jupiter.api.Test + +import org.apache.spark.executor.RayDPExecutor +import org.apache.spark.rpc.{RpcEndpointRef, RpcEnv} + +class RayAppMasterTest { + private class TestApplicationInfo extends ApplicationInfo( + 0L, + "app-test", + ApplicationDescription("test", 1, Some(1), 512, 1.0, + Command("", Map.empty, Seq.empty, Seq.empty, Seq.empty)), + new Date(0L), + null) { + var stoppedActors = 0 + + override protected def exitExecutorActor(handle: ActorHandle[RayDPExecutor]): Unit = { + stoppedActors += 1 + } + } + + @Test + def consecutiveRestartsRetainTheOriginalOwnerHandle(): Unit = { + withApp { (app, endpoint) => + val owner = app.actorSlots("0").handle + var previousId = "0" + for (executorId <- Seq("1", "2")) { + assertTrue(app.kill(previousId, shutdownActor = false)) + val reply = endpoint.askSync[AddPendingRestartedExecutorReply]( + RequestAddPendingRestartedExecutor("0")) + + assertEquals(Some(executorId), reply.newExecutorId) + assertSame(owner, app.actorSlots("0").handle) + assertSame(owner, app.getExecutorHandler(executorId).get) + assertEquals(Some(executorId), app.actorSlots("0").currentExecutorId) + assertEquals(Some(previousId), app.actorSlots("0").previousExecutorId) + assertEquals(Set(executorId), app.executors.keySet.toSet) + assertEquals(Map(executorId -> "0"), app.getRestartedExecutors) + assertEquals(1, app.actorSlots.size) + assertEquals(0, app.numActorsToAdd) + assertTrue(app.registerExecutor(executorId)) + previousId = executorId + } + assertEquals(0, app.stoppedActors) + } + } + + @Test + def removedSlotRejectsLateRestartWithoutAllocatingAnExecutorId(): Unit = { + withApp { (app, endpoint) => + assertTrue(app.kill("0", shutdownActor = false)) + assertTrue(app.kill("0", shutdownActor = true)) + app.updateDesiredExecutors(0) + + val reply = endpoint.askSync[AddPendingRestartedExecutorReply]( + RequestAddPendingRestartedExecutor("0")) + + assertEquals(None, reply.newExecutorId) + assertTrue(app.actorSlots.isEmpty) + assertTrue(app.executors.isEmpty) + assertTrue(app.executorIdToActorId.isEmpty) + assertEquals(1, app.stoppedActors) + assertEquals(0, app.numActorsToAdd) + assertEquals(1, app.getNextExecutorId()) + } + } + + @Test + def unknownSlotRejectsRestartWithoutChangingExistingSlots(): Unit = { + withApp { (app, endpoint) => + val owner = app.actorSlots("0").handle + val reply = endpoint.askSync[AddPendingRestartedExecutorReply]( + RequestAddPendingRestartedExecutor("unknown")) + + assertEquals(None, reply.newExecutorId) + assertSame(owner, app.actorSlots("0").handle) + assertEquals(Set("0"), app.actorSlots.keySet.toSet) + assertEquals(Set("0"), app.executors.keySet.toSet) + assertEquals(Map("0" -> "0"), app.executorIdToActorId.toMap) + assertEquals(1, app.getNextExecutorId()) + } + } + + private def withApp(test: (TestApplicationInfo, RpcEndpointRef) => Unit): Unit = { + val master = new RayAppMaster("127.0.0.1", 0, "") + val envField = classOf[RayAppMaster].getDeclaredField("rpcEnv") + envField.setAccessible(true) + val env = envField.get(master).asInstanceOf[RpcEnv] + try { + val endpoint = new master.RayAppMasterEndpoint(env) + val appField = endpoint.getClass.getDeclaredFields + .find(_.getName.endsWith("appInfo")).get + appField.setAccessible(true) + val app = new TestApplicationInfo + val executorId = app.getNextExecutorId().toString + app.addPendingRegisterExecutor(executorId, executorId, ownerHandle(), 1, 512) + assertTrue(app.registerExecutor(executorId)) + appField.set(endpoint, app) + + // Exercise the real RPC handler without starting Ray. A restart must reuse the stored + // owner handle rather than require a named lookup or invoke an actor method. + test(app, env.setupEndpoint("owner-handle-test", endpoint)) + } finally { + master.stop() + env.awaitTermination() + } + } + + private def ownerHandle(): ActorHandle[RayDPExecutor] = { + val handler = new InvocationHandler { + override def invoke(proxy: AnyRef, method: Method, args: Array[AnyRef]): AnyRef = { + throw new UnsupportedOperationException("Unexpected actor call: " + method.getName) + } + } + Proxy.newProxyInstance(classOf[ActorHandle[_]].getClassLoader, + Array[Class[_]](classOf[ActorHandle[_]]), handler) + .asInstanceOf[ActorHandle[RayDPExecutor]] + } +} From 790ac43a2dfc17f5bb0dfcadbe1def22babb5912 Mon Sep 17 00:00:00 2001 From: epsilonwang Date: Fri, 16 Oct 2026 09:48:51 +0000 Subject: [PATCH 2/2] test: explain executor restart regression coverage Signed-off-by: epsilonwang --- .../spark/deploy/raydp/RayAppMasterTest.scala | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/core/raydp-main/src/test/scala/org/apache/spark/deploy/raydp/RayAppMasterTest.scala b/core/raydp-main/src/test/scala/org/apache/spark/deploy/raydp/RayAppMasterTest.scala index 644f7a05..641aa8e0 100644 --- a/core/raydp-main/src/test/scala/org/apache/spark/deploy/raydp/RayAppMasterTest.scala +++ b/core/raydp-main/src/test/scala/org/apache/spark/deploy/raydp/RayAppMasterTest.scala @@ -27,7 +27,12 @@ import org.junit.jupiter.api.Test import org.apache.spark.executor.RayDPExecutor import org.apache.spark.rpc.{RpcEndpointRef, RpcEnv} +/** + * Exercises restart registration through Spark RPC without starting a Ray runtime. + * These tests check handle identity and slot bookkeeping, not native reference counting or GC. + */ class RayAppMasterTest { + // Keep the real slot-removal logic, but record shutdowns instead of submitting Ray actor tasks. private class TestApplicationInfo extends ApplicationInfo( 0L, "app-test", @@ -47,12 +52,15 @@ class RayAppMasterTest { withApp { (app, endpoint) => val owner = app.actorSlots("0").handle var previousId = "0" + // Each disconnect retires a Spark generation; the same Ray actor keeps slot "0". for (executorId <- Seq("1", "2")) { assertTrue(app.kill(previousId, shutdownActor = false)) val reply = endpoint.askSync[AddPendingRestartedExecutorReply]( RequestAddPendingRestartedExecutor("0")) assertEquals(Some(executorId), reply.newExecutorId) + // Matching actor IDs is insufficient: a lookup could return another wrapper for that + // actor. The AppMaster must keep the exact original owner handle reachable. assertSame(owner, app.actorSlots("0").handle) assertSame(owner, app.getExecutorHandler(executorId).get) assertEquals(Some(executorId), app.actorSlots("0").currentExecutorId) @@ -71,6 +79,7 @@ class RayAppMasterTest { @Test def removedSlotRejectsLateRestartWithoutAllocatingAnExecutorId(): Unit = { withApp { (app, endpoint) => + // Model scale-down after a disconnect: retire the generation, then release its slot. assertTrue(app.kill("0", shutdownActor = false)) assertTrue(app.kill("0", shutdownActor = true)) app.updateDesiredExecutors(0) @@ -84,6 +93,7 @@ class RayAppMasterTest { assertTrue(app.executorIdToActorId.isEmpty) assertEquals(1, app.stoppedActors) assertEquals(0, app.numActorsToAdd) + // Only the initial executor consumed ID 0; a rejected request must not consume ID 1. assertEquals(1, app.getNextExecutorId()) } } @@ -92,6 +102,7 @@ class RayAppMasterTest { def unknownSlotRejectsRestartWithoutChangingExistingSlots(): Unit = { withApp { (app, endpoint) => val owner = app.actorSlots("0").handle + // An unrelated actor must not acquire a slot or disturb the existing executor. val reply = endpoint.askSync[AddPendingRestartedExecutorReply]( RequestAddPendingRestartedExecutor("unknown")) @@ -100,6 +111,7 @@ class RayAppMasterTest { assertEquals(Set("0"), app.actorSlots.keySet.toSet) assertEquals(Set("0"), app.executors.keySet.toSet) assertEquals(Map("0" -> "0"), app.executorIdToActorId.toMap) + // Only the initial executor consumed ID 0; a rejected request must not consume ID 1. assertEquals(1, app.getNextExecutorId()) } } @@ -111,6 +123,8 @@ class RayAppMasterTest { val env = envField.get(master).asInstanceOf[RpcEnv] try { val endpoint = new master.RayAppMasterEndpoint(env) + // Seed the private application state directly: normal application registration would + // schedule real Ray actors, which is outside the scope of these RPC tests. val appField = endpoint.getClass.getDeclaredFields .find(_.getName.endsWith("appInfo")).get appField.setAccessible(true) @@ -130,6 +144,8 @@ class RayAppMasterTest { } private def ownerHandle(): ActorHandle[RayDPExecutor] = { + // This handle is only an identity marker. Any actor invocation fails immediately, so an + // accidental dependency on native Ray operations cannot silently pass the test. val handler = new InvocationHandler { override def invoke(proxy: AnyRef, method: Method, args: Array[AnyRef]): AnyRef = { throw new UnsupportedOperationException("Unexpected actor call: " + method.getName)