This is an automated email from the ASF dual-hosted git repository.
LuciferYang 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 1b5cab533ea4 [SPARK-58674][CORE] Assign a name to the error condition
`_LEGACY_ERROR_TEMP_3021-3023,3026,3029`
1b5cab533ea4 is described below
commit 1b5cab533ea4b3a7c400c0d9267b9b5e02f8fbaa
Author: YangJie <[email protected]>
AuthorDate: Mon Aug 10 12:53:32 2026 +0800
[SPARK-58674][CORE] Assign a name to the error condition
`_LEGACY_ERROR_TEMP_3021-3023,3026,3029`
### 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](https://issues.apache.org/jira/browse/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
[...]
- **`_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 c [...]
- **`_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 th [...]
- `_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 si [...]
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` [...]
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 4.8)
Closes #57879 from LuciferYang/assign-name-legacy-3021-3037.
Authored-by: YangJie <[email protected]>
Signed-off-by: yangjie01 <[email protected]>
(cherry picked from commit bf03f518fdd65b32bff32adf9104dc0456e43a12)
Signed-off-by: yangjie01 <[email protected]>
---
.../src/main/resources/error/error-conditions.json | 54 ++++++++++++----------
.../org/apache/spark/errors/SparkCoreErrors.scala | 18 ++++----
.../org/apache/spark/scheduler/TaskInfo.scala | 3 +-
.../apache/spark/scheduler/DAGSchedulerSuite.scala | 16 +++++++
.../org/apache/spark/scheduler/TaskInfoSuite.scala | 53 +++++++++++++++++++++
.../spark/scheduler/TaskSchedulerImplSuite.scala | 24 ++++++++++
.../KubernetesClusterSchedulerBackendSuite.scala | 18 +++++++-
7 files changed, 150 insertions(+), 36 deletions(-)
diff --git a/common/utils/src/main/resources/error/error-conditions.json
b/common/utils/src/main/resources/error/error-conditions.json
index a575551475be..3813e0b546d2 100644
--- a/common/utils/src/main/resources/error/error-conditions.json
+++ b/common/utils/src/main/resources/error/error-conditions.json
@@ -1048,6 +1048,12 @@
],
"sqlState" : "42000"
},
+ "CLUSTER_MANAGER_APPLICATION_FAILURE" : {
+ "message" : [
+ "Exiting due to an error reported by the cluster manager: <message>"
+ ],
+ "sqlState" : "56000"
+ },
"CODEC_NOT_AVAILABLE" : {
"message" : [
"The codec <codecName> is not available."
@@ -6846,6 +6852,24 @@
],
"sqlState" : "21000"
},
+ "SCHEDULER_BACKEND_SHUTDOWN_FAILED" : {
+ "message" : [
+ "Failed to shut down the scheduler backend while the application was
terminating:"
+ ],
+ "subClass" : {
+ "DRIVER_ENDPOINT" : {
+ "message" : [
+ "the driver endpoint did not stop cleanly."
+ ]
+ },
+ "EXECUTORS" : {
+ "message" : [
+ "the driver could not confirm that the executors were asked to shut
down, so the later shutdown steps were skipped."
+ ]
+ }
+ },
+ "sqlState" : "58030"
+ },
"SCHEMA_ALREADY_EXISTS" : {
"message" : [
"Cannot create schema <schemaName> because it already exists.",
@@ -8350,6 +8374,11 @@
"The row shall have a schema to get an index of the field
<fieldName>."
]
},
+ "TASK_NOT_FINISHED" : {
+ "message" : [
+ "The task has not finished yet, so its duration is not available."
+ ]
+ },
"WITHOUT_SUGGESTION" : {
"message" : [
""
@@ -11449,36 +11478,11 @@
"Checkpoint dir must be specified."
]
},
- "_LEGACY_ERROR_TEMP_3021" : {
- "message" : [
- "Error asking standalone scheduler to shut down executors"
- ]
- },
- "_LEGACY_ERROR_TEMP_3022" : {
- "message" : [
- "Error stopping standalone scheduler's driver endpoint"
- ]
- },
- "_LEGACY_ERROR_TEMP_3023" : {
- "message" : [
- "Can't run submitMapStage on RDD with 0 partitions"
- ]
- },
- "_LEGACY_ERROR_TEMP_3026" : {
- "message" : [
- "duration() called on unfinished task"
- ]
- },
"_LEGACY_ERROR_TEMP_3028" : {
"message" : [
"<errorMsg>"
]
},
- "_LEGACY_ERROR_TEMP_3029" : {
- "message" : [
- "Exiting due to error from cluster scheduler: <message>"
- ]
- },
"_LEGACY_ERROR_TEMP_3033" : {
"message" : [
"Unable to register with external shuffle server due to : <message>"
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 873254e491f2..a114246c8978 100644
--- a/core/src/main/scala/org/apache/spark/errors/SparkCoreErrors.scala
+++ b/core/src/main/scala/org/apache/spark/errors/SparkCoreErrors.scala
@@ -201,13 +201,15 @@ private[spark] object SparkCoreErrors {
def askStandaloneSchedulerToShutDownExecutorsError(e: Exception): Throwable
= {
new SparkException(
- errorClass = "_LEGACY_ERROR_TEMP_3021", messageParameters = Map.empty,
cause = e
+ errorClass = "SCHEDULER_BACKEND_SHUTDOWN_FAILED.EXECUTORS",
+ messageParameters = Map.empty, cause = e
)
}
def stopStandaloneSchedulerDriverEndpointError(e: Exception): Throwable = {
new SparkException(
- errorClass = "_LEGACY_ERROR_TEMP_3022", messageParameters = Map.empty,
cause = e
+ errorClass = "SCHEDULER_BACKEND_SHUTDOWN_FAILED.DRIVER_ENDPOINT",
+ messageParameters = Map.empty, cause = e
)
}
@@ -242,9 +244,7 @@ private[spark] object SparkCoreErrors {
}
def cannotRunSubmitMapStageOnZeroPartitionRDDError(): Throwable = {
- new SparkException(
- errorClass = "_LEGACY_ERROR_TEMP_3023", messageParameters = Map.empty,
cause = null
- )
+ SparkException.internalError("Can't run submitMapStage on RDD with 0
partitions.")
}
def accessNonExistentAccumulatorError(id: Long): Throwable = {
@@ -260,8 +260,10 @@ private[spark] object SparkCoreErrors {
new TimeoutException(s"The event queue is not empty after $timeoutMillis
ms.")
}
- def durationCalledOnUnfinishedTaskError(): Throwable = {
- new SparkUnsupportedOperationException("_LEGACY_ERROR_TEMP_3026")
+ def durationCalledOnUnfinishedTaskError(className: String, methodName:
String): Throwable = {
+ new SparkUnsupportedOperationException(
+ errorClass = "UNSUPPORTED_CALL.TASK_NOT_FINISHED",
+ messageParameters = Map("className" -> className, "methodName" ->
methodName))
}
def sparkError(errorMsg: String): Throwable = {
@@ -274,7 +276,7 @@ private[spark] object SparkCoreErrors {
def clusterSchedulerError(message: String): Throwable = {
new SparkException(
- errorClass = "_LEGACY_ERROR_TEMP_3029",
+ errorClass = "CLUSTER_MANAGER_APPLICATION_FAILURE",
messageParameters = Map("message" -> message),
cause = null
)
diff --git a/core/src/main/scala/org/apache/spark/scheduler/TaskInfo.scala
b/core/src/main/scala/org/apache/spark/scheduler/TaskInfo.scala
index 9ed95870d240..1eca39c26fcc 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/TaskInfo.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/TaskInfo.scala
@@ -149,7 +149,8 @@ class TaskInfo(
def duration: Long = {
if (!finished) {
- throw SparkCoreErrors.durationCalledOnUnfinishedTaskError()
+ throw SparkCoreErrors.durationCalledOnUnfinishedTaskError(
+ classOf[TaskInfo].getName, "duration")
} else {
finishTime - launchTime
}
diff --git
a/core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala
b/core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala
index a0428d8d8bd4..7d51ce58449c 100644
--- a/core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala
+++ b/core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala
@@ -8067,6 +8067,22 @@ class DAGSchedulerSuite extends SparkFunSuite with
TempLocalSparkContext with Ti
s"expected reason to mention '$reasonSubstring', got: $msg")
}
+ test("submitMapStage on an RDD with 0 partitions reports an internal error")
{
+ val shuffleMapRdd = new MyRDD(sc, 0, Nil)
+ val shuffleDep = new ShuffleDependency(shuffleMapRdd, new
HashPartitioner(1))
+ // `ShuffleExchangeExec` guards this case, so reaching it means an
internal invariant broke.
+ checkError(
+ exception = intercept[SparkException] {
+ scheduler.submitMapStage(shuffleDep, (_: MapOutputStatistics) => (),
CallSite("", ""),
+ new Properties())
+ },
+ condition = "INTERNAL_ERROR",
+ sqlState = Some("XX000"),
+ parameters = scala.collection.immutable.Map(
+ "message" -> "Can't run submitMapStage on RDD with 0 partitions."))
+ assertDataStructuresEmpty()
+ }
+
}
class DAGSchedulerAbortStageOffSuite extends DAGSchedulerSuite {
diff --git a/core/src/test/scala/org/apache/spark/scheduler/TaskInfoSuite.scala
b/core/src/test/scala/org/apache/spark/scheduler/TaskInfoSuite.scala
new file mode 100644
index 000000000000..7467f25f933d
--- /dev/null
+++ b/core/src/test/scala/org/apache/spark/scheduler/TaskInfoSuite.scala
@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.scheduler
+
+import org.apache.spark.{SparkFunSuite, SparkUnsupportedOperationException,
TaskState}
+
+class TaskInfoSuite extends SparkFunSuite {
+
+ private def newTaskInfo(): TaskInfo = new TaskInfo(
+ taskId = 42L,
+ index = 0,
+ attemptNumber = 0,
+ partitionId = 0,
+ launchTime = 100L,
+ executorId = "exec-1",
+ host = "host-1",
+ taskLocality = TaskLocality.PROCESS_LOCAL,
+ speculative = false)
+
+ test("duration is not available before the task finishes") {
+ val info = newTaskInfo()
+ checkError(
+ exception = intercept[SparkUnsupportedOperationException] {
+ info.duration
+ },
+ condition = "UNSUPPORTED_CALL.TASK_NOT_FINISHED",
+ sqlState = Some("0A000"),
+ parameters = Map(
+ "className" -> "org.apache.spark.scheduler.TaskInfo",
+ "methodName" -> "duration"))
+ }
+
+ test("duration is available once the task has finished") {
+ val info = newTaskInfo()
+ info.markFinished(TaskState.FINISHED, 300L)
+ assert(info.duration === 200L)
+ }
+}
diff --git
a/core/src/test/scala/org/apache/spark/scheduler/TaskSchedulerImplSuite.scala
b/core/src/test/scala/org/apache/spark/scheduler/TaskSchedulerImplSuite.scala
index 81782df283ee..118730d93168 100644
---
a/core/src/test/scala/org/apache/spark/scheduler/TaskSchedulerImplSuite.scala
+++
b/core/src/test/scala/org/apache/spark/scheduler/TaskSchedulerImplSuite.scala
@@ -2866,4 +2866,28 @@ class TaskSchedulerImplSuite extends SparkFunSuite with
LocalSparkContext
assert(taskScheduler.outstandingTasksForOtherWorkInProfile(defaultRp,
Set(0)) === 0)
}
+ test("error() with no active task sets reports a cluster manager failure") {
+ val taskScheduler = setupScheduler()
+ // No task set has been submitted, so `error` cannot abort anything and
throws instead.
+ checkError(
+ exception = intercept[SparkException] {
+ taskScheduler.error("Master removed our application: FAILED")
+ },
+ condition = "CLUSTER_MANAGER_APPLICATION_FAILURE",
+ sqlState = Some("56000"),
+ parameters = Map("message" -> "Master removed our application: FAILED"))
+ }
+
+ test("error() with an active task set aborts it instead of throwing") {
+ val taskScheduler = setupScheduler()
+ taskScheduler.submitTasks(FakeTask.createTaskSet(1))
+ val tsm = taskScheduler.taskSetManagerForAttempt(0, 0).get
+ taskScheduler.error("Master removed our application: FAILED")
+ assert(tsm.isZombie)
+ // The zombie flag cannot show whether the reason survived the trip, so
assert that the
+ // DAGScheduler was notified with the message verbatim.
+ assert(failedTaskSet)
+ assert(failedTaskSetReason === "Master removed our application: FAILED")
+ }
+
}
diff --git
a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterSchedulerBackendSuite.scala
b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterSchedulerBackendSuite.scala
index cf172eb096d4..97228871942a 100644
---
a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterSchedulerBackendSuite.scala
+++
b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterSchedulerBackendSuite.scala
@@ -31,14 +31,14 @@ import org.mockito.ArgumentMatchers.{any, eq => mockitoEq}
import org.mockito.Mockito.{atLeastOnce, mock, never, spy, verify, when}
import org.scalatest.BeforeAndAfter
-import org.apache.spark.{SparkConf, SparkContext, SparkEnv, SparkFunSuite}
+import org.apache.spark.{SparkConf, SparkContext, SparkEnv, SparkException,
SparkFunSuite}
import org.apache.spark.deploy.k8s.Config._
import org.apache.spark.deploy.k8s.Constants._
import org.apache.spark.deploy.k8s.Fabric8Aliases._
import org.apache.spark.resource.{ResourceProfile, ResourceProfileManager}
import org.apache.spark.rpc.{RpcCallContext, RpcEndpoint, RpcEndpointRef,
RpcEnv}
import org.apache.spark.scheduler.{ExecutorKilled, ExecutorLossReason,
LiveListenerBus, TaskSchedulerImpl}
-import
org.apache.spark.scheduler.cluster.CoarseGrainedClusterMessages.{RegisterExecutor,
RemoveExecutor, StopDriver}
+import
org.apache.spark.scheduler.cluster.CoarseGrainedClusterMessages.{RegisterExecutor,
RemoveExecutor, StopDriver, StopExecutors}
import org.apache.spark.scheduler.cluster.CoarseGrainedSchedulerBackend
import
org.apache.spark.scheduler.cluster.k8s.ExecutorLifecycleTestUtils.TEST_SPARK_APP_ID
@@ -306,6 +306,20 @@ class KubernetesClusterSchedulerBackendSuite extends
SparkFunSuite with BeforeAn
verify(kubernetesClient).close()
}
+ test("stopExecutors() reports a failed StopExecutors RPC as
SCHEDULER_BACKEND_SHUTDOWN_FAILED") {
+ val rpcFailure = new RuntimeException("StopExecutors timed out")
+
when(driverEndpointRef.askSync[Boolean](StopExecutors)).thenThrow(rpcFailure)
+ val e = intercept[SparkException] {
+ schedulerBackendUnderTest.stopExecutors()
+ }
+ checkError(
+ exception = e,
+ condition = "SCHEDULER_BACKEND_SHUTDOWN_FAILED.EXECUTORS",
+ sqlState = Some("58030"),
+ parameters = Map.empty[String, String])
+ assert(e.getCause === rpcFailure)
+ }
+
test("SPARK-34469: Ignore RegisterExecutor when SparkContext is stopped") {
when(sc.isStopped).thenReturn(true)
val endpoint = schedulerBackendUnderTest.createDriverEndpoint()
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
