sunchao commented on code in PR #57525:
URL: https://github.com/apache/spark/pull/57525#discussion_r3666977630
##########
core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala:
##########
@@ -104,7 +104,7 @@ private[spark] class CoarseGrainedExecutorBackend(
driver = Some(ref)
env.executorBackend = Option(this)
ref.ask[Boolean](RegisterExecutor(executorId, self, hostname, cores,
extractLogUrls,
- extractAttributes, _resources, resourceProfile.id))
+ extractAttributes, _resources, resourceProfile.id, env.conf.getAppId))
Review Comment:
[P1] Validate identity before returning bootstrap credentials
Passing the launch application ID only in `RegisterExecutor` is too late.
Before this point the executor has already sent `RetrieveSparkAppConfig`, whose
request carries only the resource-profile ID, and `DriverEndpoint`
unconditionally returns its Spark properties, optional I/O-encryption key, and
Hadoop delegation credentials. In the port-reuse scenario an executor for
application A can therefore fetch application B's bootstrap secrets before B
rejects registration; the executor installs the returned delegation tokens
before reaching this sender. This is reachable with the default
`spark.authenticate=false`. Please carry `arguments.appId` in
`RetrieveSparkAppConfig` and reject a mismatch before constructing
`SparkAppConfig`, while keeping this registration check for a swap after config
fetch.
##########
core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala:
##########
@@ -247,8 +247,12 @@ class CoarseGrainedSchedulerBackend(scheduler:
TaskSchedulerImpl, val rpcEnv: Rp
override def receiveAndReply(context: RpcCallContext):
PartialFunction[Any, Unit] = {
case RegisterExecutor(executorId, executorRef, hostname, cores, logUrls,
- attributes, resources, resourceProfileId) =>
- if (executorDataMap.contains(executorId)) {
+ attributes, resources, resourceProfileId, appId) =>
+ if (Option(appId).exists(_ != scheduler.applicationId())) {
Review Comment:
[P1] Wait for the authoritative application ID before comparing
`scheduler.applicationId()` does not imply that the cluster-manager ID is
initialized. In standalone mode, `StandaloneSchedulerBackend.applicationId()`
returns an unrelated generated `spark-application-*` fallback until the
asynchronous `connected(appId)` callback publishes the master's ID. The master
sends `RegisteredApplication` and immediately calls `schedule()` over a
different RPC path, so a legitimately launched executor can reach this handler
first carrying the real master-assigned ID and be rejected against the
fallback; those exits consume the executor retry budget and can fail
applications configured with a low `spark.deploy.maxExecutorRetries`. Please
expose an explicit, safely published ID-ready state and defer/retry
registration until it is ready instead of comparing against the fallback.
##########
core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedClusterMessage.scala:
##########
@@ -69,7 +69,8 @@ private[spark] object CoarseGrainedClusterMessages {
logUrls: Map[String, String],
attributes: Map[String, String],
resources: Map[String, ResourceInformation],
- resourceProfileId: Int)
+ resourceProfileId: Int,
+ appId: String = null)
Review Comment:
[P2] Preserve the existing RegisterExecutor ABI
The default argument only helps newly compiled Scala call sites. Adding a
ninth case-class field changes the generated constructor, `apply`, and
`unapply` signatures: Java or precompiled external backends using the
eight-argument form can fail linkage, and eight-field pattern matches no longer
compile (the required `BlockManagerSuite` edits demonstrate the extractor
break). A live external example is [Armada's eight-field
match](https://github.com/armadaproject/armada-spark/blob/master/src/main/scala/org/apache/spark/scheduler/cluster/armada/ArmadaClusterManagerBackend.scala#L674-L679).
Since this PR explicitly intends compatibility with custom/external cluster
managers, please retain the old `RegisterExecutor` shape and introduce a
versioned identity-carrying message or equivalent compatibility path.
##########
core/src/test/scala/org/apache/spark/scheduler/CoarseGrainedSchedulerBackendSuite.scala:
##########
@@ -606,6 +606,143 @@ class CoarseGrainedSchedulerBackendSuite extends
SparkFunSuite with LocalSparkCo
assert(mockEndpointRef.decommissionReceived)
}
+ test("SPARK-58322: reject RegisterExecutor with mismatched app ID") {
+ val conf = new SparkConf()
+ .setMaster("local-cluster[0, 3, 1024]")
+ .setAppName("test")
+ sc = new SparkContext(conf)
+ val backend =
sc.schedulerBackend.asInstanceOf[CoarseGrainedSchedulerBackend]
+ val mockEndpointRef = mock[RpcEndpointRef]
+ val mockAddress = mock[RpcAddress]
+
+ val ex = intercept[SparkException] {
+ backend.driverEndpoint.askSync[Boolean](
+ RegisterExecutor("1", mockEndpointRef, mockAddress.host, 1, Map.empty,
Map.empty,
+ Map.empty, ResourceProfile.DEFAULT_RESOURCE_PROFILE_ID,
"wrong-app-id"))
+ }
+ assert(ex.getCause.getMessage.contains("Executor app ID wrong-app-id does
not match"))
+ }
+
+ test("SPARK-58322: accept RegisterExecutor with matching app ID") {
+ val conf = new SparkConf()
+ .setMaster("local-cluster[0, 3, 1024]")
+ .setAppName("test")
+ sc = new SparkContext(conf)
+ val backend =
sc.schedulerBackend.asInstanceOf[CoarseGrainedSchedulerBackend]
+ val mockEndpointRef = mock[RpcEndpointRef]
+ val mockAddress = mock[RpcAddress]
+
+ val result = backend.driverEndpoint.askSync[Boolean](
+ RegisterExecutor("1", mockEndpointRef, mockAddress.host, 1, Map.empty,
Map.empty,
+ Map.empty, ResourceProfile.DEFAULT_RESOURCE_PROFILE_ID,
sc.applicationId))
+ assert(result)
+ }
+
+ test("SPARK-58322: accept RegisterExecutor with null app ID for backward
compatibility") {
+ val conf = new SparkConf()
+ .setMaster("local-cluster[0, 3, 1024]")
+ .setAppName("test")
+ sc = new SparkContext(conf)
+ val backend =
sc.schedulerBackend.asInstanceOf[CoarseGrainedSchedulerBackend]
+ val mockEndpointRef = mock[RpcEndpointRef]
+ val mockAddress = mock[RpcAddress]
+
+ val result = backend.driverEndpoint.askSync[Boolean](
+ RegisterExecutor("1", mockEndpointRef, mockAddress.host, 1, Map.empty,
Map.empty,
+ Map.empty, ResourceProfile.DEFAULT_RESOURCE_PROFILE_ID, null))
+ assert(result)
+ }
+
+ test("SPARK-58322: reject RegisterExecutor after driver swap between config
fetch and " +
+ "registration") {
+ // Simulate the port-reuse scenario: executor fetches config from driver
A, then driver A
+ // dies and releases its RPC port, driver B binds the same address, and
the executor sends
+ // RegisterExecutor to driver B. The app ID carried by the executor (from
driver A) will not
+ // match driver B's applicationId, so registration must be rejected.
+ val driverAAppId = "app-driver-A"
+
+ // Set up a fake "driver A" endpoint that responds to
RetrieveSparkAppConfig.
+ val driverRpcEnv = RpcEnv.create("test-driverA", "localhost", 0, new
SparkConf(),
+ new SecurityManager(new SparkConf()), clientMode = false)
+ try {
+ driverRpcEnv.setupEndpoint("fake-driverA", new RpcEndpoint {
+ override val rpcEnv: RpcEnv = driverRpcEnv
+ override def receiveAndReply(context: RpcCallContext):
PartialFunction[Any, Unit] = {
+ case RetrieveSparkAppConfig(_) =>
+ context.reply(SparkAppConfig(
+ Seq("spark.app.id" -> driverAAppId),
+ None, None, ResourceProfile.getOrCreateDefaultProfile(new
SparkConf()), None))
+ }
+ })
+
+ // Executor fetches config from driver A.
+ val driverARef =
driverRpcEnv.setupEndpointRefByURI("spark://fake-driverA@localhost:" +
+ driverRpcEnv.address.port)
+ val cfg = driverARef.askSync[SparkAppConfig](
+ RetrieveSparkAppConfig(ResourceProfile.DEFAULT_RESOURCE_PROFILE_ID))
+ val fetchedAppId = cfg.sparkProperties.find(_._1 ==
"spark.app.id").map(_._2).orNull
+ assert(fetchedAppId == driverAAppId)
+
+ // Now create driver B (a real SparkContext with a different
applicationId).
+ val conf = new SparkConf()
+ .setMaster("local-cluster[0, 3, 1024]")
+ .setAppName("test")
+ sc = new SparkContext(conf)
+ val backend =
sc.schedulerBackend.asInstanceOf[CoarseGrainedSchedulerBackend]
+ assert(sc.applicationId != driverAAppId)
+
+ // Executor sends RegisterExecutor to driver B with the app ID from
driver A.
+ val mockEndpointRef = mock[RpcEndpointRef]
+ val mockAddress = mock[RpcAddress]
+ val ex = intercept[SparkException] {
+ backend.driverEndpoint.askSync[Boolean](
+ RegisterExecutor("1", mockEndpointRef, mockAddress.host, 1,
Map.empty, Map.empty,
+ Map.empty, ResourceProfile.DEFAULT_RESOURCE_PROFILE_ID,
fetchedAppId))
Review Comment:
[P2] Exercise the production sender in this regression
This test reads a synthetic config and then manually places `fetchedAppId`
into `RegisterExecutor`; it never runs `CoarseGrainedExecutorBackend.onStart`
or verifies that the launch `arguments.appId` reaches the only changed
production sender. Removing or miswiring that sender still leaves every new
test green. Please drive the executor-backend bootstrap/registration path and
capture the emitted `RegisterExecutor`, asserting that it contains the launch
application ID. If the test keeps claiming an exact port-reuse regression, it
should also shut down A and have B bind A's released address.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
