[SPARK-57498][CORE] Fix ResourceProfile ID collision when custom profile is created before default

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.

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]