sunchao commented on code in PR #57525:
URL: https://github.com/apache/spark/pull/57525#discussion_r3653392212
##########
core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala:
##########
@@ -455,6 +455,7 @@ private[spark] object CoarseGrainedExecutorBackend extends
Logging {
}
val cfg =
driver.askSync[SparkAppConfig](RetrieveSparkAppConfig(arguments.resourceProfileId))
+ verifyAppId(cfg.sparkProperties, arguments.appId)
Review Comment:
[P1] Validate the application ID on the registration connection
This checks the driver identity only on `driverPropsFetcher`, but that RPC
environment is shut down immediately afterwards. The executor subsequently
creates a new `SparkEnv`, resolves `driverUrl` again in
`CoarseGrainedExecutorBackend.onStart`, and sends `RegisterExecutor` on a
different connection. `RegisterExecutor` contains no application ID, and
`DriverEndpoint` accepts an otherwise valid, previously unused executor ID.
Consequently, if the verified driver releases its port and a different
application's driver binds the same address between config retrieval and
registration, the executor still registers with the wrong application despite
this check. This reproduces the port-reuse/data-corruption scenario the PR is
intended to prevent under the default RPC authentication settings; the
Kubernetes backend has the same two-connection sequence. Please include the
executor's application ID in `RegisterExecutor` and validate it in
`DriverEndpoint`, or otherwise bind the identity che
ck to the connection used for registration. Add a regression test that swaps
the driver between config retrieval and registration.
##########
core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala:
##########
@@ -608,4 +609,23 @@ private[spark] object CoarseGrainedExecutorBackend extends
Logging {
// scalastyle:on println
System.exit(1)
}
+
+ /**
+ * Verify that the executor's app ID (from the --app-id launch argument)
matches the driver's
+ * app ID (from spark.app.id in the driver's SparkConf). A mismatch
indicates the executor has
+ * connected to the wrong driver, likely due to driver port reuse after the
original driver
+ * released its RPC port (e.g. via SparkContext.stop() following a fatal
error) while the
+ * driver process remained in a zombie state. Throw SparkException to
prevent the executor
+ * from registering to the wrong application and corrupting data.
+ */
+ private[spark] def verifyAppId(
+ sparkProperties: Seq[(String, String)],
+ executorAppId: String): Unit = {
+ sparkProperties.find(_._1 == "spark.app.id").map(_._2).foreach {
driverAppId =>
Review Comment:
[P1] Do not accept an application ID missing from the driver's cached
configuration
`DriverEndpoint.sparkProperties` is an immutable `lazy val`, initialized by
the first `RetrieveSparkAppConfig` request. However, `SparkContext` calls
`_taskScheduler.start()` before `_conf.set("spark.app.id", _applicationId)`,
and the standalone scheduler can launch an executor during that startup. If the
first executor fetches its config in this window, the cached driver properties
omit `spark.app.id` for the entire application lifetime. This
`.find(...).foreach` then silently succeeds for every later executor, including
one belonging to another application, so the proposed protection is permanently
disabled. The new `appId verification no-op when spark.app.id absent` test
actually codifies this failure. Please obtain the finalized application ID from
an authoritative driver/scheduler source and reject an unverifiable identity,
ideally in the registration handler; add a regression test that fetches
configuration before the driver publishes its application ID.
--
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]
