From de5b6eb6d6d03a304dd78b959897fffd9d80d3ab Mon Sep 17 00:00:00 2001 From: ybapat Date: Mon, 27 Jul 2026 10:49:29 -0400 Subject: [PATCH] [SPARK-57498][CORE] Fix ResourceProfile ID collision when custom profile is created before default Start nextProfileId at 1 instead of 0 to avoid colliding with DEFAULT_RESOURCE_PROFILE_ID (0). Previously a custom ResourceProfile created before getOrCreateDefaultProfile() was called would receive ID 0, indistinguishable from the default profile. Also reset nextProfileId in clearDefaultProfile() for clean test isolation. Co-Authored-By: Claude Sonnet 4.6 --- .../spark/resource/ResourceProfile.scala | 8 ++++++- .../spark/resource/ResourceProfileSuite.scala | 22 +++++++++++++++++++ 2 files changed, 29 insertions(+), 1 deletion(-) diff --git a/core/src/main/scala/org/apache/spark/resource/ResourceProfile.scala b/core/src/main/scala/org/apache/spark/resource/ResourceProfile.scala index e397e80fe5738..f012d461666fe 100644 --- a/core/src/main/scala/org/apache/spark/resource/ResourceProfile.scala +++ b/core/src/main/scala/org/apache/spark/resource/ResourceProfile.scala @@ -371,7 +371,8 @@ object ResourceProfile extends Logging { val UNKNOWN_RESOURCE_PROFILE_ID = -1 val DEFAULT_RESOURCE_PROFILE_ID = 0 - private lazy val nextProfileId = new AtomicInteger(0) + // Start at 1: ID 0 is reserved for DEFAULT_RESOURCE_PROFILE_ID. + private lazy val nextProfileId = new AtomicInteger(1) private val DEFAULT_PROFILE_LOCK = new Object() // The default resource profile uses the application level configs. @@ -473,6 +474,11 @@ object ResourceProfile extends Logging { } } + // for testing only + private[spark] def resetNextProfileIdForTesting(): Unit = { + nextProfileId.set(1) + } + /* * Get the number of cpus per task if its set in the profile, otherwise return the * cpus per task for the default profile. diff --git a/core/src/test/scala/org/apache/spark/resource/ResourceProfileSuite.scala b/core/src/test/scala/org/apache/spark/resource/ResourceProfileSuite.scala index 8f92c474a5250..35942f7fdab83 100644 --- a/core/src/test/scala/org/apache/spark/resource/ResourceProfileSuite.scala +++ b/core/src/test/scala/org/apache/spark/resource/ResourceProfileSuite.scala @@ -38,6 +38,7 @@ class ResourceProfileSuite extends SparkFunSuite with MockitoSugar { override def afterEach(): Unit = { try { ResourceProfile.clearDefaultProfile() + ResourceProfile.resetNextProfileIdForTesting() } finally { super.afterEach() } @@ -498,6 +499,27 @@ class ResourceProfileSuite extends SparkFunSuite with MockitoSugar { new ResourceProfileBuilder().require(treqs).build() } + test("custom ResourceProfile id must not collide with DEFAULT_RESOURCE_PROFILE_ID") { + // Create a custom ResourceProfile *before* the default profile is initialised. + // Prior to the fix, getNextProfileId started at 0 so the first custom profile + // received ID 0, which is identical to DEFAULT_RESOURCE_PROFILE_ID. + val customRp = new ResourceProfileBuilder() + .require(new ExecutorResourceRequests().cores(2).memory("2048")) + .require(new TaskResourceRequests().cpus(1)) + .build() + + val defaultRp = ResourceProfile.getOrCreateDefaultProfile(new SparkConf) + + assert(customRp.id != ResourceProfile.DEFAULT_RESOURCE_PROFILE_ID, + s"Custom profile id (${customRp.id}) must not equal DEFAULT_RESOURCE_PROFILE_ID " + + s"(${ResourceProfile.DEFAULT_RESOURCE_PROFILE_ID})") + assert(customRp.id != defaultRp.id, + s"Custom profile id (${customRp.id}) must not equal the default profile id (${defaultRp.id})") + assert(defaultRp.id == ResourceProfile.DEFAULT_RESOURCE_PROFILE_ID, + s"Default profile id (${defaultRp.id}) must equal DEFAULT_RESOURCE_PROFILE_ID " + + s"(${ResourceProfile.DEFAULT_RESOURCE_PROFILE_ID})") + } + private def withMockSparkEnv(conf: SparkConf)(f: => Unit): Unit = { val previousEnv = SparkEnv.get val mockEnv = mock[SparkEnv]