[SPARK-41246][CORE] Fail fast on RDD id overflow; parse negative block ids as safety net

core/src/main/scala/org/apache/spark/SparkContext.scala

@@ -2903,8 +2903,34 @@ class SparkContext(config: SparkConf) extends Logging {
 
   private val nextRddId = new AtomicInteger(0)
 
-  /** Register a new RDD, returning its RDD ID */
-  private[spark] def newRddId(): Int = nextRddId.getAndIncrement()
+  /**
+   * Testing helper: set the next value that [[newRddId]] will return.
+   */
+  private[spark] def setNextRddIdForTesting(value: Int): Unit = nextRddId.set(value)
+
+  /**
+   * Register a new RDD, returning its RDD ID.
+   *
+   * Fails if the 32-bit counter would wrap to a negative value. Continuing with
+   * wrapped ids can break BlockManager, UI, and other id-keyed state.
+   * [[org.apache.spark.storage.BlockId]] still parses negative RDD names as a
+   * safety net for any in-flight or pre-upgrade cached blocks.
+   */
+  private[spark] def newRddId(): Int = {
+    val id = nextRddId.getAndIncrement()
+    if (id < 0) {
+      // Stay pegged so subsequent allocations keep failing clearly.
+      nextRddId.set(Int.MinValue)
+      throw new SparkException(
+        "RDD id counter overflowed Int.MaxValue (" + Int.MaxValue +
+          "). This application has created too many RDDs; restart it.")
+    }
+    if (id == Int.MaxValue) {
+      logWarning("Allocated the last valid RDD id (Int.MaxValue). " +
+        "Further RDD creation will fail.")
+    }
+    id
+  }
 
   /**
    * Registers listeners specified in spark.extraListeners, then starts the listener bus.

core/src/main/scala/org/apache/spark/storage/BlockId.scala

@@ -263,7 +263,11 @@ case class CacheId(sessionUUID: String, hash: String) extends BlockId {
 
 @DeveloperApi
 object BlockId {
-  val RDD = "rdd_([0-9]+)_([0-9]+)".r
+  // Safety net: newRddId() fails fast before minting negative ids, but names like
+  // rdd_-1330910599_36 may still appear from blocks cached before upgrade or from
+  // tests. Accept an optional minus so BlockId.apply does not throw
+  // UnrecognizedBlockId for those names.
+  val RDD = "rdd_(-?[0-9]+)_([0-9]+)".r
   val SHUFFLE = "shuffle_([0-9]+)_([0-9]+)_([0-9]+)".r
   val SHUFFLE_BATCH = "shuffle_([0-9]+)_([0-9]+)_([0-9]+)_([0-9]+)".r
   val SHUFFLE_DATA = "shuffle_([0-9]+)_([0-9]+)_([0-9]+).data".r

core/src/test/scala/org/apache/spark/SparkContextSuite.scala

@@ -1521,6 +1521,19 @@ class SparkContextSuite extends SparkFunSuite with LocalSparkContext with Eventu
     sc = new SparkContext(conf)
     assert(sc.env.memoryManager.maxOffHeapStorageMemory > 0)
   }
+
+  test("SPARK-41246: fail-fast on RDD id overflow") {
+    val conf = new SparkConf().setAppName("test").setMaster("local[1]")
+    sc = new SparkContext(conf)
+    sc.setNextRddIdForTesting(Int.MaxValue)
+    val last = sc.parallelize(Seq(1), 1)
+    assert(last.id === Int.MaxValue)
+    val err = intercept[SparkException] {
+      sc.parallelize(Seq(2), 1)
+    }
+    assert(err.getMessage.contains("Int.MaxValue"))
+    assert(err.getMessage.contains("overflowed"))
+  }
 }
 
 object SparkContextSuite {

core/src/test/scala/org/apache/spark/storage/BlockIdSuite.scala

@@ -51,6 +51,16 @@ class BlockIdSuite extends SparkFunSuite {
     assertSame(id, BlockId(id.toString))
   }
 
+  test("SPARK-41246: rdd with a negative id") {
+    // Safety net: fail-fast prevents minting negative ids, but BlockId must still
+    // parse names from pre-upgrade caches or tests.
+    val id = RDDBlockId(-1330910599, 36)
+    assert(id.name === "rdd_-1330910599_36")
+    assertSame(id, BlockId(id.name))
+    assertDifferent(id, RDDBlockId(1330910599, 36))
+    assertSame(RDDBlockId(Int.MinValue, 0), BlockId("rdd_-2147483648_0"))
+  }
+
   test("shuffle") {
     val id = ShuffleBlockId(1, 2, 3)
     assertSame(id, ShuffleBlockId(1, 2, 3))