This is an automated email from the ASF dual-hosted git repository.
LuciferYang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new a2e47babfc2c [SPARK-58702][CORE] Assign a name to the error condition
`_LEGACY_ERROR_TEMP_3033,3035-3037`
a2e47babfc2c is described below
commit a2e47babfc2cea9e2178f1f7fbe3ae6bb0dcaa6b
Author: YangJie <[email protected]>
AuthorDate: Sat Aug 22 17:30:03 2026 +0800
[SPARK-58702][CORE] Assign a name to the error condition
`_LEGACY_ERROR_TEMP_3033,3035-3037`
### What changes were proposed in this pull request?
This PR converts the four remaining `_LEGACY_ERROR_TEMP_*` conditions in
`SparkCoreErrors`' storage and shuffle group, continuing the cleanup under
[SPARK-37935](https://issues.apache.org/jira/browse/SPARK-37935). Three get
user-facing names; one becomes an internal error.
| Legacy | Builder | Now | SQLSTATE |
|---|---|---|---|
| `_LEGACY_ERROR_TEMP_3033` |
`unableToRegisterWithExternalShuffleServerError` |
`UNABLE_TO_REGISTER_WITH_EXTERNAL_SHUFFLE_SERVICE` | 58030 |
| `_3035` | `shuffleBlockMigrationNotSupportedError` (renamed) |
`SHUFFLE_BLOCK_MIGRATION_NOT_SUPPORTED` | 0A000 |
| `_3036` | `failToStoreBlockOnBlockManagerError` |
`INTERNAL_ERROR_STORAGE` (entry deleted) | XX000 |
| `_3037` | `localBlockDataNotFoundError` (renamed) |
`LOCAL_BLOCK_DATA_NOT_FOUND` | 58030 |
This is the last batch of the `_3021-3042` cluster. `_3028` is left out on
purpose: its builder is a bare `<errorMsg>` passthrough whose single call site
both duplicates the `taskSet.abort` on the line above it and has its throw
swallowed by `Inbox.safelyCall`, so converting it honestly needs a design
decision of its own rather than a rename.
### Why `_3036` gets no name
It is thrown from the `StreamCallbackWithID.onComplete` that
`BlockManager.putBlockDataAsStream` returns, when the block store declines a
replica. The condition cannot survive to a reader: `TransportRequestHandler`
wraps the exception in a plain `IOException` and answers `RpcFailure` carrying
only `stackTraceToString(e)`, and the sender rebuilds that string as a
`RuntimeException`. By the time anything logs it, the condition name, SQLSTATE
and message parameters are gone — a name woul [...]
Its category is `STORAGE`, not the `NETWORK` that
`NettyBlockRpcServer.receive` uses, because the category tracks the package of
the throw site — here `org.apache.spark.storage`. Every other `category`
argument in `SparkCoreErrors` is `STORAGE`.
### Reachability, per named condition
- **`_3033` — operator/environment failure that propagates.**
`registerWithExternalShuffleServer` retries
`spark.shuffle.registration.maxAttempts` times (default 3); the `case
NonFatal(e)` arm normally fires only on the last attempt. Nothing catches it up
the chain, so the executor fails to construct. Actionable by whoever runs the
shuffle service.
- **`_3035` — third-party extension code.** Reached only when a third-party
`BlockingShuffleManager` supplies a `ShuffleBlockResolver` that does not mix in
`MigratableResolver`, and that executor receives a migrated shuffle block
during decommissioning. In-tree managers all resolve to
`IndexShuffleBlockResolver`, which does implement it; a manager providing no
resolver at all fails earlier with a different error. Per the precedent of
`CANNOT_LOAD_CATALOG`, the extension author is the [...]
- **`_3037` — environment failure, and the throw is the SPARK-15736
recovery.** A `DiskStore` file vanished under a read lock.
`handleLocalReadFailure` de-registers the block so its unavailability reaches
the driver, then fails this attempt. It surfaces in the driver's task-failure
output at least once per lost block.
### Does this PR introduce _any_ user-facing change?
Yes, to error messages — no API change. Converting any legacy condition
also adds the `[CONDITION] ` prefix and the ` SQLSTATE: xxxxx` suffix, since
`SparkThrowableHelper.formatErrorMessage` suppresses both only for
`_LEGACY_ERROR_`-prefixed names.
Two of the four had defects beyond the rename:
- **`_3035` rendered an identity hash.** It interpolated a
`ShuffleBlockResolver` object, and neither that trait nor
`IndexShuffleBlockResolver` overrides `toString`, so the message read
`org.apache.spark.shuffle.Foo1a2b3c4d`. It now passes `getClass.getName`.
- **`_3033` had a stray space before its colon** (`due to : <message>`),
and now falls back to `e.toString` when the cause carries no message —
otherwise a message-less throwable rendered the literal `null`.
The three new messages deliberately avoid claiming more than the code
guarantees, which took several passes to get right:
- **No deictic machine reference.** "this executor" is wrong wherever the
text crosses an RPC boundary and is read on the peer — `_3037` can be thrown
while serving a remote read, and `BlockManager` also runs on the driver.
`_3035` says "the receiving executor" rather than "this executor" because the
party that reads it for diagnosis is the migration *source*.
- **A requirement, not an asserted cause.** `_3035` is raised from a `catch
case e: ClassCastException`, and the cast lives in a lazy val — so once
initialized, a CCE from inside a third-party `putShuffleBlockAsStream` reaches
the same handler. The message says the resolver *must implement*
`MigratableResolver` instead of asserting it does not, and the CCE is now
carried as the cause so a mislabelled one stays diagnosable.
- **No recovery promise.** `_3037` says a later read *must* fetch from a
replica or recompute (a necessity, not an availability claim). A sealed
`localCheckpoint` block has had its lineage cut and can never be recomputed, so
promising recovery would be false there. Its tense is past for the state at
read time, since `removeBlock` has already run by the time the message is built.
`_3036`'s rendered text changes from `Failure while trying to store block
<blockId> on <blockManagerId>.` to the internal-error form, and now carries a
diagnostic hint, since by the reachability argument above that text is the
whole diagnostic value the error has. The hint names storage memory rather than
disk space: the only way to reach this throw is a memory-only level whose
`acquireStorageMemory` fails, because `saveToDiskStore()` returns `Unit` and a
failed disk write throws an ` [...]
### How was this patch tested?
Three existing assertions in `BlockManagerSuite` are tightened to
`checkError`, none of which a grep for the condition token would have caught.
Two of them cover the newly named conditions on the read and migration paths:
- "we reject putting blocks when we have the wrong shuffle resolver" pinned
only a substring of the message prose, leaving the condition and SQLSTATE
unchecked — and it is what forced an edit here, since this PR rewords that
message. It now asserts the condition, SQLSTATE `0A000`, both parameters, and
that the `ClassCastException` survives as the cause.
- `testReadWithLossOfOnDiskFiles` (driving "remove block if a read fails
due to missing DiskStore files (SPARK-15736)") was a bare
`intercept[SparkException]` that checked nothing but the exception type. It now
asserts `LOCAL_BLOCK_DATA_NOT_FOUND` with SQLSTATE `58030`.
`_3033`'s `SPARK-39647` assertion is now a `checkError` on the condition,
SQLSTATE `58030` and the `message` parameter, matched as a regex on the
deterministic `java.util.concurrent.TimeoutException: ` prefix, since the rest
of that value is a Guava internal that varies run to run. Its other two
assertions (`SPARK-20640` ×2) still read the cause text out of `getMessage`,
which together with that parameter assertion is why `message` is kept alongside
`cause` rather than dropped as redu [...]
One deliberate deviation: `error/README.md` says a parameter value should
not itself be a message, and `UNABLE_TO_REGISTER_WITH_EXTERNAL_SHUFFLE_SERVICE`
passes the cause's text as `<message>`. Dropping it would mean rewriting the
two `getMessage.contains(...)` assertions above against `getCause`, since
`SparkException.getMessage` does not include the cause. It is also a
well-trodden deviation rather than a new one: 27 other non-legacy conditions
carry a `<message>` placeholder, and ` [...]
Ran locally on top of the current master: `SparkThrowableSuite` (34 tests)
and `BlockManagerSuite` (130 tests), all passing; `core/compile` clean; the
JSON regenerated with `SPARK_GENERATE_GOLDEN_FILES=1` produced no extra diff.
Each new assertion was verified to actually bite by temporarily breaking what
it checks — each SQLSTATE assertion and the cause — and watching the test go
red.
On the SQLSTATE choices: 58030 is nominally "I/O error" but is already
Spark's de-facto system-failure code, carried by 12 conditions before this PR,
several of which are not literal I/O; `UNABLE_TO_FETCH_HIVE_TABLES` is the
closest precedent for `_3033`. `0A000` for `_3035` follows
`PIPELINED_SHUFFLE_UNSUPPORTED` in the same subsystem — the resolver loads fine
and merely lacks a capability, which is why `46103` ("unresolved class name",
used by `CANNOT_LOAD_CATALOG`) does not fit. `e [...]
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 4.8)
Closes #57923 from LuciferYang/assign-name-legacy-3033-3037.
Authored-by: YangJie <[email protected]>
Signed-off-by: yangjie01 <[email protected]>
---
.../src/main/resources/error/error-conditions.json | 38 ++++++++++------------
.../org/apache/spark/errors/SparkCoreErrors.scala | 31 ++++++++----------
.../org/apache/spark/storage/BlockManager.scala | 8 ++---
.../apache/spark/storage/BlockManagerSuite.scala | 37 ++++++++++++++++-----
4 files changed, 65 insertions(+), 49 deletions(-)
diff --git a/common/utils/src/main/resources/error/error-conditions.json
b/common/utils/src/main/resources/error/error-conditions.json
index aa492d647f12..c642e0be953a 100644
--- a/common/utils/src/main/resources/error/error-conditions.json
+++ b/common/utils/src/main/resources/error/error-conditions.json
@@ -5766,6 +5766,12 @@
],
"sqlState" : "42K03"
},
+ "LOCAL_BLOCK_DATA_NOT_FOUND" : {
+ "message" : [
+ "Block <blockId> had block metadata but its data was missing from both
the memory and disk stores. The block has been removed, so a later read must
fetch it from another replica or recompute it."
+ ],
+ "sqlState" : "58030"
+ },
"LOCAL_MUST_WITH_SCHEMA_FILE" : {
"message" : [
"LOCAL must be used together with the schema of `file`, but got:
`<actualSchema>`."
@@ -7050,6 +7056,12 @@
],
"sqlState" : "42K05"
},
+ "SHUFFLE_BLOCK_MIGRATION_NOT_SUPPORTED" : {
+ "message" : [
+ "Shuffle block <blockId> cannot be migrated to the receiving executor:
its shuffle block resolver <resolverClass> must implement MigratableResolver.
See the cause for details."
+ ],
+ "sqlState" : "0A000"
+ },
"SKETCH_INVALID_LG_NOM_ENTRIES" : {
"message" : [
"Invalid call to <function>; the `lgNomEntries` value must be between
<min> and <max>, inclusive: <value>."
@@ -8265,6 +8277,12 @@
],
"sqlState" : "42KD9"
},
+ "UNABLE_TO_REGISTER_WITH_EXTERNAL_SHUFFLE_SERVICE" : {
+ "message" : [
+ "Unable to register with the external shuffle service: <message>"
+ ],
+ "sqlState" : "58030"
+ },
"UNBOUND_SQL_PARAMETER" : {
"message" : [
"Found the unbound parameter: <name>. Please, fix `args` and provide a
mapping of the parameter to either a SQL literal or collection constructor
functions such as `map()`, `array()`, `struct()`."
@@ -11625,26 +11643,6 @@
"<errorMsg>"
]
},
- "_LEGACY_ERROR_TEMP_3033" : {
- "message" : [
- "Unable to register with external shuffle server due to : <message>"
- ]
- },
- "_LEGACY_ERROR_TEMP_3035" : {
- "message" : [
- "Unexpected shuffle block <blockId> with unsupported shuffle resolver
<shuffleBlockResolver>"
- ]
- },
- "_LEGACY_ERROR_TEMP_3036" : {
- "message" : [
- "Failure while trying to store block <blockId> on <blockManagerId>."
- ]
- },
- "_LEGACY_ERROR_TEMP_3037" : {
- "message" : [
- "Block <blockId> was not found even though it's read-locked"
- ]
- },
"_LEGACY_ERROR_TEMP_3052" : {
"message" : [
"Unexpected resolved action: <other>"
diff --git a/core/src/main/scala/org/apache/spark/errors/SparkCoreErrors.scala
b/core/src/main/scala/org/apache/spark/errors/SparkCoreErrors.scala
index 442d4623c4a9..d23469da5b4e 100644
--- a/core/src/main/scala/org/apache/spark/errors/SparkCoreErrors.scala
+++ b/core/src/main/scala/org/apache/spark/errors/SparkCoreErrors.scala
@@ -323,8 +323,8 @@ private[spark] object SparkCoreErrors {
def unableToRegisterWithExternalShuffleServerError(e: Throwable): Throwable
= {
new SparkException(
- errorClass = "_LEGACY_ERROR_TEMP_3033",
- messageParameters = Map("message" -> e.getMessage),
+ errorClass = "UNABLE_TO_REGISTER_WITH_EXTERNAL_SHUFFLE_SERVICE",
+ messageParameters = Map("message" ->
Option(e.getMessage).getOrElse(e.toString)),
cause = e
)
}
@@ -333,35 +333,32 @@ private[spark] object SparkCoreErrors {
SparkException.internalError("Error occurred while waiting for async.
reregistration.", e)
}
- def unexpectedShuffleBlockWithUnsupportedResolverError(
+ def shuffleBlockMigrationNotSupportedError(
+ blockId: BlockId,
shuffleBlockResolver: ShuffleBlockResolver,
- blockId: BlockId): Throwable = {
+ e: Throwable): Throwable = {
new SparkException(
- errorClass = "_LEGACY_ERROR_TEMP_3035",
+ errorClass = "SHUFFLE_BLOCK_MIGRATION_NOT_SUPPORTED",
messageParameters = Map(
"blockId" -> s"$blockId",
- "shuffleBlockResolver" -> s"$shuffleBlockResolver"
+ "resolverClass" -> shuffleBlockResolver.getClass.getName
),
- cause = null
+ cause = e
)
}
def failToStoreBlockOnBlockManagerError(
blockManagerId: BlockManagerId,
blockId: BlockId): Throwable = {
- new SparkException(
- errorClass = "_LEGACY_ERROR_TEMP_3036",
- messageParameters = Map(
- "blockId" -> s"$blockId",
- "blockManagerId" -> s"$blockManagerId"
- ),
- cause = null
- )
+ SparkException.internalError(
+ s"Failed to store block $blockId on $blockManagerId. This mostly happens
when there is " +
+ "not enough storage memory for the block and its storage level has no
disk fallback.",
+ category = "STORAGE")
}
- def readLockedBlockNotFoundError(blockId: BlockId): Throwable = {
+ def localBlockDataNotFoundError(blockId: BlockId): Throwable = {
new SparkException(
- errorClass = "_LEGACY_ERROR_TEMP_3037",
+ errorClass = "LOCAL_BLOCK_DATA_NOT_FOUND",
messageParameters = Map(
"blockId" -> s"$blockId"
),
diff --git a/core/src/main/scala/org/apache/spark/storage/BlockManager.scala
b/core/src/main/scala/org/apache/spark/storage/BlockManager.scala
index 801f4c0b7472..7267348031ed 100644
--- a/core/src/main/scala/org/apache/spark/storage/BlockManager.scala
+++ b/core/src/main/scala/org/apache/spark/storage/BlockManager.scala
@@ -948,9 +948,9 @@ private[spark] class BlockManager(
try {
return migratableResolver.putShuffleBlockAsStream(blockId,
serializerManager)
} catch {
- case _: ClassCastException =>
- throw
SparkCoreErrors.unexpectedShuffleBlockWithUnsupportedResolverError(
- shuffleBlockResolver, blockId)
+ case e: ClassCastException =>
+ throw SparkCoreErrors.shuffleBlockMigrationNotSupportedError(
+ blockId, shuffleBlockResolver, e)
}
}
logDebug(s"Putting regular block ${blockId}")
@@ -1128,7 +1128,7 @@ private[spark] class BlockManager(
releaseLock(blockId)
// Remove the missing block so that its unavailability is reported to the
driver
removeBlock(blockId)
- throw SparkCoreErrors.readLockedBlockNotFoundError(blockId)
+ throw SparkCoreErrors.localBlockDataNotFoundError(blockId)
}
private def isIORelatedException(t: Throwable): Boolean =
diff --git
a/core/src/test/scala/org/apache/spark/storage/BlockManagerSuite.scala
b/core/src/test/scala/org/apache/spark/storage/BlockManagerSuite.scala
index e2e2552270f7..f60e7800ab49 100644
--- a/core/src/test/scala/org/apache/spark/storage/BlockManagerSuite.scala
+++ b/core/src/test/scala/org/apache/spark/storage/BlockManagerSuite.scala
@@ -1781,9 +1781,13 @@ class BlockManagerSuite extends SparkFunSuite with
Matchers with PrivateMethodTe
// Because the BlockManager's metadata claims that the block exists (i.e.
that it's present
// in at least one store), the read attempts to read it and fails when the
on-disk file is
// missing.
- intercept[SparkException] {
- readMethod(store)
- }
+ checkError(
+ exception = intercept[SparkException] {
+ readMethod(store)
+ },
+ condition = "LOCAL_BLOCK_DATA_NOT_FOUND",
+ sqlState = Some("58030"),
+ parameters = Map("blockId" -> "test_blockId"))
// Subsequent read attempts will succeed; the block isn't present but we
return an expected
// "block not found" response rather than a fatal error:
assert(readMethod(store).isEmpty)
@@ -2065,7 +2069,16 @@ class BlockManagerSuite extends SparkFunSuite with
Matchers with PrivateMethodTe
val exception = intercept[SparkException] {
bm.putBlockDataAsStream(shuffleBlockId, StorageLevel.DISK_ONLY,
ClassTag(message.getClass))
}
- assert(exception.getMessage.contains("unsupported shuffle resolver"))
+ checkError(
+ exception = exception,
+ condition = "SHUFFLE_BLOCK_MIGRATION_NOT_SUPPORTED",
+ sqlState = Some("0A000"),
+ parameters = Map(
+ "blockId" -> shuffleBlockId.toString,
+ "resolverClass" -> badShuffleResolver.getClass.getName))
+ // The `ClassCastException` is kept as the cause. The `try` only covers
building the callback,
+ // so this pins the cast failure itself rather than anything the resolver
does while writing.
+ assert(exception.getCause.isInstanceOf[ClassCastException])
}
test("SPARK-54796: putBlockDataAsStream throws
ShuffleManagerNotInitializedException " +
@@ -2401,10 +2414,18 @@ class BlockManagerSuite extends SparkFunSuite with
Matchers with PrivateMethodTe
conf.set(SHUFFLE_SERVICE_PORT.key, shufflePort.toString)
conf.set(SHUFFLE_REGISTRATION_TIMEOUT.key, "40")
conf.set(SHUFFLE_REGISTRATION_MAX_ATTEMPTS.key, "1")
- val e = intercept[SparkException] {
- makeBlockManager(8000, "timeoutExec")
- }.getMessage
- assert(e.contains("TimeoutException"))
+ // `matchPVals` because the parameter is the text of the
`RuntimeException` that
+ // `TransportClient.sendRpcSync` wraps Guava's `TimeoutException` in
(`cause.toString`).
+ // Only the class-name prefix of that text is stable; the rest is a
Guava internal that
+ // varies run to run (e.g. the future's identity hash and a
scheduling-delay clause).
+ checkError(
+ exception = intercept[SparkException] {
+ makeBlockManager(8000, "timeoutExec")
+ },
+ condition = "UNABLE_TO_REGISTER_WITH_EXTERNAL_SHUFFLE_SERVICE",
+ sqlState = Some("58030"),
+ parameters = Map("message" ->
"(?s)java\\.util\\.concurrent\\.TimeoutException: .*"),
+ matchPVals = true)
verify(master, times(0))
.registerBlockManager(mc.any(), mc.any(), mc.any(), mc.any(),
mc.any(), mc.any())
server.close()
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
