Skip to content

[SPARK-58322][CORE] Verify executor app ID matches driver at config fetch #57525

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wangyum wants to merge 1 commit into
base: master
Choose a base branch
from
Open

[SPARK-58322][CORE] Verify executor app ID matches driver at config fetch #57525

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 20 additions & 0 deletions core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -455,6 +455,7 @@ private[spark] object CoarseGrainedExecutorBackend extends Logging {
}

val cfg = driver.askSync[SparkAppConfig](RetrieveSparkAppConfig(arguments.resourceProfileId))
verifyAppId(cfg.sparkProperties, arguments.appId)

@sunchao sunchao Jul 26, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[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 check to the connection used for registration. Add a regression test that swaps the driver between config retrieval and registration.

val props = cfg.sparkProperties ++ Seq[(String, String)](("spark.app.id", arguments.appId))
fetcher.shutdown()

Expand Down Expand Up @@ -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 =>

@sunchao sunchao Jul 26, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[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.

if (driverAppId != executorAppId) {
throw new SparkException(s"Executor app ID $executorAppId does not match driver app ID " +
s"$driverAppId. This likely means the executor connected to the wrong driver. Exiting.")
}
}
}
}
18 changes: 18 additions & 0 deletions core/src/test/scala/org/apache/spark/executor/CoarseGrainedExecutorBackendSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -626,6 +626,24 @@ class CoarseGrainedExecutorBackendSuite extends SparkFunSuite
SparkEnv.set(mockEnv)
mockEnv
}

test("appId verification passes when app IDs match") {
val props = Seq("spark.app.id" -> "app-123")
CoarseGrainedExecutorBackend.verifyAppId(props, "app-123")
}

test("appId verification throws when app IDs mismatch") {
val props = Seq("spark.app.id" -> "app-new")
val ex = intercept[SparkException] {
CoarseGrainedExecutorBackend.verifyAppId(props, "app-old")
}
assert(ex.getMessage.contains("Executor app ID app-old does not match driver app ID app-new."))
}

test("appId verification no-op when spark.app.id absent from driver properties") {
val props = Seq("spark.executor.cores" -> "2")
CoarseGrainedExecutorBackend.verifyAppId(props, "app-old")
}
}

private class TestFatalErrorPlugin extends SparkPlugin {
Expand Down
1 change: 1 addition & 0 deletions ...ore/src/main/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesExecutorBackend.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -95,6 +95,7 @@ private[spark] object KubernetesExecutorBackend extends Logging {
}

val cfg = driver.askSync[SparkAppConfig](RetrieveSparkAppConfig(arguments.resourceProfileId))
CoarseGrainedExecutorBackend.verifyAppId(cfg.sparkProperties, arguments.appId)
val props = cfg.sparkProperties ++ Seq[(String, String)](("spark.app.id", arguments.appId))
val execId: String = arguments.executorId match {
case null | "EXECID" | "" =>
Expand Down