This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 933a9caf0114 [SPARK-41246][CORE] Fail fast on RDD id overflow; parse
negative block ids as safety net
933a9caf0114 is described below
commit 933a9caf0114cb9fa9840ba4f7a97d0c7fa1a080
Author: Giri Prakash reddy <[email protected]>
AuthorDate: Wed Aug 12 08:19:04 2026 +0800
[SPARK-41246][CORE] Fail fast on RDD id overflow; parse negative block ids
as safety net
### What changes were proposed in this pull request?
1. **Fail-fast (primary fix):** `SparkContext.newRddId` throws if the
32-bit counter would wrap to a negative value after `Int.MaxValue`, so
long-running apps no longer mint negative RDD ids that break BlockManager /
other id-keyed state.
2. **Parse safety net:** `BlockId` RDD regex accepts an optional minus
(`rdd_(-?[0-9]+)_...`) so names like `rdd_-1330910599_36` from pre-upgrade
caches or tests still parse instead of throwing `UnrecognizedBlockId`.
Under fail-fast, newly created RDDs never get negative ids; the parser
change is defense in depth.
### Why are the changes needed?
After more than `Int.MaxValue` RDDs, `AtomicInteger` wraps to negatives.
Cached partitions then use names such as `rdd_-1330910599_36`, and
`BlockId.apply` failed with `UnrecognizedBlockId` (e.g. on `UpdateBlockInfo`).
Parsing negatives alone would let the app continue with wrapped ids and cover
up overflow elsewhere. Fail-fast makes overflow an explicit error; the regex
remains for in-flight / old negative names.
### Does this PR introduce _any_ user-facing change?
Yes. Creating an RDD after `Int.MaxValue` throws a clear `SparkException`
instead of wrapping. Restart the application to reset the counter. Positive
block-id parsing is unchanged. Java `int id = rdd.id()` still compiles.
### How was this patch tested?
- `BlockIdSuite`: negative RDD block name round-trip (safety net)
- `SparkContextSuite`: fail-fast when counter is at `Int.MaxValue`
```
build/sbt "core/testOnly org.apache.spark.storage.BlockIdSuite"
build/sbt "core/testOnly org.apache.spark.SparkContextSuite -- -z
SPARK-41246"
```
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Cursor Agent
Closes #57893 from giriprakash/SPARK-41246-negative-rdd-block-id.
Authored-by: Giri Prakash reddy <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
(cherry picked from commit 3df71625ce243d4f76099cab3d4fa7b52f13006c)
Signed-off-by: Wenchen Fan <[email protected]>
---
.../main/scala/org/apache/spark/SparkContext.scala | 30 ++++++++++++++++++++--
.../scala/org/apache/spark/storage/BlockId.scala | 6 ++++-
.../scala/org/apache/spark/SparkContextSuite.scala | 13 ++++++++++
.../org/apache/spark/storage/BlockIdSuite.scala | 10 ++++++++
4 files changed, 56 insertions(+), 3 deletions(-)
diff --git a/core/src/main/scala/org/apache/spark/SparkContext.scala
b/core/src/main/scala/org/apache/spark/SparkContext.scala
index 355549137077..c0203aa032ba 100644
--- a/core/src/main/scala/org/apache/spark/SparkContext.scala
+++ b/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.
diff --git a/core/src/main/scala/org/apache/spark/storage/BlockId.scala
b/core/src/main/scala/org/apache/spark/storage/BlockId.scala
index 3e46a53ee082..8d44e79a223a 100644
--- a/core/src/main/scala/org/apache/spark/storage/BlockId.scala
+++ b/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
diff --git a/core/src/test/scala/org/apache/spark/SparkContextSuite.scala
b/core/src/test/scala/org/apache/spark/SparkContextSuite.scala
index 709f267de155..ef2d18f51306 100644
--- a/core/src/test/scala/org/apache/spark/SparkContextSuite.scala
+++ b/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 {
diff --git a/core/src/test/scala/org/apache/spark/storage/BlockIdSuite.scala
b/core/src/test/scala/org/apache/spark/storage/BlockIdSuite.scala
index 4b7f1fafb211..259b38c7f83d 100644
--- a/core/src/test/scala/org/apache/spark/storage/BlockIdSuite.scala
+++ b/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))
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
