[SPARK-58674][CORE] Assign a name to the error condition _LEGACY_ERROR_TEMP_3021-3023,3026,3029

[LuciferYang]

What changes were proposed in this pull request?

This PR converts five _LEGACY_ERROR_TEMP_* conditions in SparkCoreErrors -- _3021, _3022, _3023, _3026 and _3029 -- into proper error conditions, continuing the cleanup under SPARK-37935. Four get user-facing names; one is an unreachable defensive check and becomes an internal error. Note the set is not a contiguous range: _3028 sits inside it and is deliberately left out (see below).

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 (inside this PR's numeric span, hence the non-contiguous title), plus _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 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)

[dongjoon-hyun]

+1, LGTM. I verified the changes against the codebase:

• All three JSON insertions (CLUSTER_MANAGER_APPLICATION_FAILURE, SCHEDULER_BACKEND_SHUTDOWN_FAILED, UNSUPPORTED_CALL.TASK_NOT_FINISHED) are correctly placed alphabetically, and all three SQLSTATEs exist in error-states.json with reasonable precedents (56000 already used by CHECKPOINT_RDD_BLOCK_ID_NOT_FOUND, 58030 by 11 conditions).
• durationCalledOnUnfinishedTaskError has exactly one caller, so the signature change is safe, and no stray references to the five removed legacy IDs remain anywhere in the repo.
• stopExecutors() is only called from stop(), so the "later shutdown steps were skipped" clause in the EXECUTORS message is accurate. The base-message-ending-with-colon style has plenty of precedent (DATATYPE_MISMATCH, FAILED_JDBC, etc.).
• Exception types are unchanged, so there is no user-facing API impact and no MiMa concern.
• The new tests reuse existing suite plumbing correctly (the mocked driverEndpointRef pattern in the k8s suite, failedTaskSet/failedTaskSetReason in TaskSchedulerImplSuite).

One minor nit: the title says _LEGACY_ERROR_TEMP_3021-3029, but _3028 is intentionally left for a follow-up. Stating the exact set (3021-3023,3026,3029) would make the commit log more precise.

[uros-b]

Thank you @LuciferYang and @dongjoon-hyun!

[LuciferYang]

+1, LGTM. I verified the changes against the codebase:

• All three JSON insertions (CLUSTER_MANAGER_APPLICATION_FAILURE, SCHEDULER_BACKEND_SHUTDOWN_FAILED, UNSUPPORTED_CALL.TASK_NOT_FINISHED) are correctly placed alphabetically, and all three SQLSTATEs exist in error-states.json with reasonable precedents (56000 already used by CHECKPOINT_RDD_BLOCK_ID_NOT_FOUND, 58030 by 11 conditions).
• durationCalledOnUnfinishedTaskError has exactly one caller, so the signature change is safe, and no stray references to the five removed legacy IDs remain anywhere in the repo.
• stopExecutors() is only called from stop(), so the "later shutdown steps were skipped" clause in the EXECUTORS message is accurate. The base-message-ending-with-colon style has plenty of precedent (DATATYPE_MISMATCH, FAILED_JDBC, etc.).
• Exception types are unchanged, so there is no user-facing API impact and no MiMa concern.
• The new tests reuse existing suite plumbing correctly (the mocked driverEndpointRef pattern in the k8s suite, failedTaskSet/failedTaskSetReason in TaskSchedulerImplSuite).

One minor nit: the title says _LEGACY_ERROR_TEMP_3021-3029, but _3028 is intentionally left for a follow-up. Stating the exact set (3021-3023,3026,3029) would make the commit log more precise.

The pr title has been rewritten.

[LuciferYang]

Merge Summary:

• merged into master bf03f51
• merged into branch-4.x 1b5cab5

Posted by merge_spark_pr.py

[LuciferYang]

Thank you @dongjoon-hyun @HyukjinKwon @uros-b