wangyum commented on code in PR #57525:
URL: https://github.com/apache/spark/pull/57525#discussion_r3675994109
##########
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:
👍
--
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]
