sunchao commented on code in PR #57525:
URL: https://github.com/apache/spark/pull/57525#discussion_r3679479642
##########
core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala:
##########
@@ -140,6 +140,16 @@ class CoarseGrainedSchedulerBackend(scheduler:
TaskSchedulerImpl, val rpcEnv: Rp
ThreadUtils.newDaemonSingleThreadScheduledExecutor("cleanup-decommission-execs")
}
+ private def appIdMismatch(executorAppId: Option[String]):
Option[SparkException] = {
+ val resolvedDriverAppId = scheduler.backend.realApplicationId()
+ for {
+ execAppId <- executorAppId
+ driverAppId <- resolvedDriverAppId
Review Comment:
[P1] Do not accept an executor while the driver application ID is unavailable
When `realApplicationId()` returns `None`, this comprehension reports no
mismatch, so both `RetrieveSparkAppConfig` and `RegisterExecutor` accept any
concrete executor application ID. This is a real production window: standalone
exposes its driver endpoint before asynchronous `connected(appId)`, and YARN
client calls `super.start()` before `submitApplication()`/`bindToYarn()`. An
executor from old application A reaching reused-port driver B during that
interval can receive B's I/O-encryption key/delegation credentials and register
permanently before B's identity is published; nothing revalidates it afterward.
The new fallback test actually asserts that `"any-app-id"` is accepted. Please
defer both bootstrap and registration until the authoritative driver ID is
available, then validate the executor.
##########
core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala:
##########
@@ -140,6 +140,16 @@ class CoarseGrainedSchedulerBackend(scheduler:
TaskSchedulerImpl, val rpcEnv: Rp
ThreadUtils.newDaemonSingleThreadScheduledExecutor("cleanup-decommission-execs")
}
+ private def appIdMismatch(executorAppId: Option[String]):
Option[SparkException] = {
+ val resolvedDriverAppId = scheduler.backend.realApplicationId()
Review Comment:
[P1] Do not dereference the scheduler backend before checking executor
identity
`TaskSchedulerImpl.backend` starts as `null`, and existing mocked/custom
schedulers do not necessarily populate it, but this lookup runs unconditionally
even when the executor supplied no application ID. The current [core CI
job](https://github.com/wangyum/spark/actions/runs/30467942666/job/90635683053)
already reports five existing failures across
`CoarseGrainedSchedulerBackendSuite`, `StandaloneDynamicAllocationSuite`, and
`HeartbeatReceiverSuite`, all caused by `NullPointerException` at this line; I
also reproduced `extra resources from executor` locally. Please check whether
an executor ID exists before resolving driver identity and obtain it from the
actual enclosing backend instead of dereferencing the nullable scheduler
backlink.
##########
core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala:
##########
@@ -352,15 +365,20 @@ class CoarseGrainedSchedulerBackend(scheduler:
TaskSchedulerImpl, val rpcEnv: Rp
adjustTargetNumExecutors = false,
triggeredByExecutor = true))
- case RetrieveSparkAppConfig(resourceProfileId) =>
- val rp =
scheduler.sc.resourceProfileManager.resourceProfileFromId(resourceProfileId)
- val reply = SparkAppConfig(
- sparkProperties,
- SparkEnv.get.securityManager.getIOEncryptionKey(),
- Option(delegationTokens.get()),
- rp,
- currentLogLevel)
- context.reply(reply)
+ case RetrieveSparkAppConfig(resourceProfileId, appId) =>
+ // Validate identity before returning bootstrap credentials.
+ appIdMismatch(Option(appId)) match {
Review Comment:
[P1] Require executor identity before returning bootstrap credentials
`Option(appId)` converts the compatibility default `null` into `None`, after
which the success branch returns `SparkAppConfig` including Spark properties,
the I/O-encryption key, and Hadoop delegation credentials. This is not
hypothetical: [RayDP still sends the one-argument
request](https://github.com/ray-project/raydp/blob/58dbe26f0f6c0d26fd7e1d86477148730c1ab618/core/raydp-main/src/main/scala/org/apache/spark/executor/RayDPExecutor.scala#L209-L216),
so recompiling it against this change supplies `null` and leaves wrong-driver
credential disclosure intact; registration without an app-ID attribute is
likewise explicitly accepted. RPC authentication is disabled by default. Please
require a verified application ID for credential-bearing bootstrap requests, or
provide a compatibility path with equivalent authentication rather than
silently failing open.
##########
core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala:
##########
@@ -104,7 +104,8 @@ 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 + ("spark.app.id" -> env.conf.getAppId),
Review Comment:
[P2] Add a regression test for the production registration sender
Every added core regression manually constructs `RegisterExecutor`; none
runs `CoarseGrainedExecutorBackend.onStart` or captures the registration
message produced here. The new Kubernetes test only captures
`RetrieveSparkAppConfig` and deliberately throws from its backend factory
before registration. Therefore deleting or miswiring this `"spark.app.id"`
insertion leaves all eight new `SPARK-58322` core tests green while reopening
the post-bootstrap driver-swap vulnerability. Please add a
fake-driver/executor-backend test that captures the actual `RegisterExecutor`
emitted by `onStart` and verifies its attributes contain the launch application
ID.
##########
core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedClusterMessage.scala:
##########
@@ -29,7 +29,10 @@ private[spark] sealed trait CoarseGrainedClusterMessage
extends Serializable
private[spark] object CoarseGrainedClusterMessages {
- case class RetrieveSparkAppConfig(resourceProfileId: Int) extends
CoarseGrainedClusterMessage
+ case class RetrieveSparkAppConfig(
+ resourceProfileId: Int,
+ appId: String = null)
Review Comment:
[P2] Preserve the existing RetrieveSparkAppConfig ABI
The default argument preserves only newly compiled Scala construction
syntax. Adding this field removes the existing JVM `(int)` constructor and
companion `apply(int)`, and changes the one-field extractor; `javap` confirms
the new class exposes only `(int, String)`. A real [Spark-Proxy Java
integration directly invokes `new
RetrieveSparkAppConfig(0)`](https://github.com/Radeity/Spark-Proxy/blob/d5e9cbd243f20d30f04afa3483081dd65232416f/Dispatcher/src/main/java/org/apache/spark/java/dispatcher/DispatcherEndpoint.java#L143-L145),
which now fails even after recompilation, while previously compiled RayDP
artifacts fail linkage. This moves the earlier `RegisterExecutor` compatibility
problem to the bootstrap message. Please retain the original one-field request
and introduce a separately versioned identity-carrying request, or otherwise
preserve the old constructor, factory, and extractor.
--
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]
