LuciferYang opened a new pull request, #57879:
URL: https://github.com/apache/spark/pull/57879

   ### What changes were proposed in this pull request?
   
   This PR converts five remaining `_LEGACY_ERROR_TEMP_*` conditions in 
`SparkCoreErrors` into proper error conditions, continuing the cleanup under 
[SPARK-37935](https://issues.apache.org/jira/browse/SPARK-37935). Four get 
user-facing names; one is an unreachable defensive check and becomes an 
internal error.
   
   | Legacy | Builder | Now | SQLSTATE |
   |---|---|---|---|
   | `_LEGACY_ERROR_TEMP_3021` | 
`askStandaloneSchedulerToShutDownExecutorsError` | 
`SCHEDULER_BACKEND_SHUTDOWN_FAILED.EXECUTORS` | 58030 |
   | `_3022` | `stopStandaloneSchedulerDriverEndpointError` | 
`SCHEDULER_BACKEND_SHUTDOWN_FAILED.DRIVER_ENDPOINT` | 58030 |
   | `_3023` | `cannotRunSubmitMapStageOnZeroPartitionRDDError` | 
`INTERNAL_ERROR` (entry deleted) | XX000 |
   | `_3026` | `durationCalledOnUnfinishedTaskError` | 
`UNSUPPORTED_CALL.TASK_NOT_FINISHED` | 0A000 |
   | `_3029` | `clusterSchedulerError` | `CLUSTER_MANAGER_APPLICATION_FAILURE` 
| 56000 |
   
   `SCHEDULER_BACKEND_SHUTDOWN_FAILED` is a new umbrella: both conditions are 
the same exception type thrown at the same altitude in adjacent methods of 
`CoarseGrainedSchedulerBackend`, differing only in which RPC failed. 
`TASK_NOT_FINISHED` joins the existing `UNSUPPORTED_CALL` umbrella, whose 
message frame (`Cannot call the method "<methodName>" of the class 
"<className>".`) is exactly the shape this throw site needs.
   
   `durationCalledOnUnfinishedTaskError` now takes `className`/`methodName` 
from its call site rather than deriving them internally, matching how the other 
`UNSUPPORTED_CALL` throw sites build their parameters (`DiskBlockObjectWriter`, 
`DataTypeErrors`). The no-arg form would have silently misreported the class if 
a second caller ever appeared.
   
   The remaining conditions in the `_3021-3042` range — `_3028`, `_3033`, 
`_3035`, `_3036`, `_3037` — are left for follow-up: they are operational 
failures (a barrier stage getting partial offers, external shuffle service 
registration exhausting its retries, a replica failing to store, a `DiskStore` 
file vanishing) or reachable only through a third-party `ShuffleManager`, and 
each needs its own reachability judgement.
   
   ### Reachability, per condition
   
   - **`_3021`/`_3022` — cluster/environment failure, log-only.** Both wrap an 
`askSync` to the driver endpoint during teardown, so they fire when an RPC 
times out or the endpoint is already dead. Every path into 
`CoarseGrainedSchedulerBackend.stop()` is wrapped in 
`Utils.tryLogNonFatalError` (`SparkContext.stop` → `DAGScheduler.stop` → 
`TaskSchedulerImpl.stop`), and `KubernetesClusterSchedulerBackend.stop` wraps 
`super.stop()` explicitly, so the exception reaches a driver log and never a 
user. It is still a real operational failure rather than a Spark bug, which is 
why it gets a name instead of `INTERNAL_ERROR` — the reader is an operator 
triaging a shutdown log.
   - **`_3023` — unreachable defensive check.** `DAGScheduler.submitMapStage` 
rejects a dependency whose RDD has zero partitions. Its only production caller, 
`ShuffleExchangeExec.mapOutputStatisticsFuture`, already short-circuits with 
`if (inputRDD.getNumPartitions == 0) Future.successful(null)`, and the two 
guards test the same number: the dependency's RDD is 
`prepareShuffleDependency(inputRDD, ...)`'s output, derived through 
`mapPartitionsWithIndexInternal`, which preserves partition count. Note this is 
*not* an argument from `private[spark]` — that is a scalac-only check with no 
bytecode enforcement, so Java code or a user class declared under 
`org.apache.spark.*` can call `SparkContext.submitMapStage` directly. The 
accurate statement is that the branch is reachable only by violating the 
contract of an API whose own scaladoc says "This is currently an internal API 
only", which is what `INTERNAL_ERROR` is for.
   - **`_3026` — reachable from user code.** `TaskInfo` is `@DeveloperApi` and 
a `TaskInfo` for a *running* task is handed to listeners via 
`SparkListenerTaskStart`. Any custom listener calling `taskInfo.duration` in 
`onTaskStart` hits this. All in-tree callers are guarded or run after 
`markFinished`, which is why it never fires in the build.
   - **`_3029` — operator-facing, standalone only.** `TaskSchedulerImpl.error` 
throws when the cluster manager reports a failure and no task set is active to 
abort. Its only caller is `StandaloneSchedulerBackend.dead`, i.e. the 
standalone Master removed the application or all Masters were unresponsive. The 
condition name is deliberately deployment-agnostic because 
`TaskSchedulerImpl.error` is, but today standalone is the only producer.
   
   ### Why are the changes needed?
   
   The error-conditions 
[README](https://github.com/apache/spark/blob/master/common/utils/src/main/resources/error/README.md)
 disallows new `_LEGACY_ERROR_TEMP_*` entries and asks existing ones to be 
resolved. This clears five of them.
   
   Two of the five were also weak in their own right: `_3021`/`_3022` carried 
no indication that the failure happens during shutdown, so an operator seeing 
the log line had no way to tell whether the application's results were affected.
   
   ### Does this PR introduce _any_ user-facing change?
   
   Yes, to error messages — no API change.
   
   Converting any legacy condition changes the rendered string in two 
mechanical ways: `SparkThrowableHelper.formatErrorMessage` suppresses the 
`[CONDITION] ` prefix only for `_LEGACY_ERROR_`-prefixed names, and appends ` 
SQLSTATE: xxxxx` when a sqlState exists (legacy entries have none). Beyond that:
   
   - `_3021`: `Error asking standalone scheduler to shut down executors` → 
`Failed to shut down the scheduler backend while the application was 
terminating: the driver could not confirm that the executors were asked to shut 
down, so the later shutdown steps were skipped.` The added clause is factual: 
`stop()` calls `stopExecutors()` unguarded, so a throw there skips 
`stopTokenManager()` and the `StopDriver` ask. The wording says "could not 
confirm" rather than "could not tell" because the `StopExecutors` handler sends 
to every executor *before* replying, so a lost reply does not mean the 
executors went unnotified.
   - `_3022`: `Error stopping standalone scheduler's driver endpoint` → `... : 
the driver endpoint did not stop cleanly.`
   - `_3026`: `duration() called on unfinished task` → `Cannot call the method 
"duration" of the class "org.apache.spark.scheduler.TaskInfo". The task has not 
finished yet, so its duration is not available.`
   - `_3029`: `Exiting due to error from cluster scheduler: <message>` → 
`Exiting due to an error reported by the cluster manager: <message>`. This one 
is genuinely user-visible in logs, so log-scraping on the old text would need 
updating.
   - `_3023` renders as an internal error. Note this does **not** change any 
job-failure message shape: the throw happens on the driver during 
`submitMapStage`, before `eventProcessLoop.post`, so 
`DAGScheduler.abortStage`'s `isInternalError` filter — which only applies to 
exceptions arriving from task-set failures — is not involved.
   
   ### How was this patch tested?
   
   New assertions, four of which fail against the old code (the condition name 
and the SQLSTATE both differ, since legacy entries carry no sqlState):
   
   - `TaskInfoSuite` (new file) — `checkError` on 
`UNSUPPORTED_CALL.TASK_NOT_FINISHED` including SQLSTATE and both message 
parameters, plus a companion case asserting `duration` still works after 
`markFinished`.
   - `TaskSchedulerImplSuite` — `error()` with no active task set asserts 
`CLUSTER_MANAGER_APPLICATION_FAILURE`; a companion case covers the other branch 
(with an active task set it aborts rather than throwing) and asserts the reason 
reaches the `DAGScheduler` intact.
   - `KubernetesClusterSchedulerBackendSuite` — a failed `StopExecutors` RPC 
asserts `SCHEDULER_BACKEND_SHUTDOWN_FAILED.EXECUTORS` and that the original 
exception survives as the cause.
   - `DAGSchedulerSuite` — `submitMapStage` on a zero-partition RDD asserts 
`INTERNAL_ERROR`, so the converted branch is pinned rather than merely deleted 
from the JSON.
   
   Coverage caveat worth a reviewer's attention: the only assertion on 
`SCHEDULER_BACKEND_SHUTDOWN_FAILED` lives in the `kubernetes` module and so 
runs only under `-Pkubernetes`. Both throw sites are in `core`, but `core`'s 
`CoarseGrainedSchedulerBackendSuite` uses a real `local-cluster` context with 
no way to make `askSync` fail, whereas the k8s suite already has the mocked 
`RpcEndpointRef` plumbing. The `DRIVER_ENDPOINT` subclass has no assertion for 
the same reason — its throw site sits inside `stop()`, which the k8s backend 
wraps in `Utils.tryLogNonFatalError`.
   
   Ran locally: `SparkThrowableSuite`, `TaskInfoSuite`, 
`TaskSchedulerImplSuite`, `DAGSchedulerSuite` (367 tests) and 
`KubernetesClusterSchedulerBackendSuite` (12 tests), all passing; 
`core/compile`, `core/Test/compile` and `kubernetes/Test/compile` clean. The 
JSON was regenerated with `SPARK_GENERATE_GOLDEN_FILES=1` and produced no diff. 
Each new SQLSTATE assertion was verified to actually bite by temporarily 
setting a wrong value and watching the test go red.
   
   On the SQLSTATE choice for `SCHEDULER_BACKEND_SHUTDOWN_FAILED`: 58030 is 
nominally "I/O error", and a failed RPC is not literally I/O. It was chosen 
because 58030 is already Spark's de-facto system-operation-failure code, 
carrying 11 conditions of which several are not literal I/O 
(`FAILED_UPDATE_VIEW_SCHEMA`, `FAILED_TO_CREATE_PLAN_FOR_DIRECT_QUERY`, 
`CANNOT_RESTORE_PERMISSIONS_FOR_PATH`), whereas `58000` ("System error") and 
the `08xxx` connection class exist in `error-states.json` but have zero users 
in `error-conditions.json` — introducing the first would set a new precedent in 
a cleanup PR. Happy to switch if a committer prefers otherwise.
   
   ### Was this patch authored or co-authored using generative AI tooling?
   
   Generated-by: Claude Code (Opus 4.8)
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: [email protected]

For queries about this service, please contact Infrastructure at:
[email protected]


---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]

Reply via email to