You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by do...@apache.org on 2026/07/22 17:24:37 UTC
(spark) branch branch-4.x updated: [SPARK-57465][CORE] Do not count RejectedExecutionException from a shutting-down executor toward task failures
This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun 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 372f76c92dd1 [SPARK-57465][CORE] Do not count RejectedExecutionException from a shutting-down executor toward task failures
372f76c92dd1 is described below
commit 372f76c92dd12ce3b655bc19d7afd510eb01450c
Author: Anupam Yadav <an...@amazon.com>
AuthorDate: Wed Jul 22 10:24:16 2026 -0700
[SPARK-57465][CORE] Do not count RejectedExecutionException from a shutting-down executor toward task failures
### What changes were proposed in this pull request?
Add a dedicated `ExecutorShutdownFailure` TaskFailedReason with `countTowardsTaskFailures = false`. When the executor thread pool rejects a task with `RejectedExecutionException` while the executor is shutting down (gated on the `executorShutdown` flag), this reason is reported so the attempt is not counted toward `spark.task.maxFailures` and the task is rescheduled elsewhere.
### Why are the changes needed?
Tasks launched onto an executor whose thread pool is shutting down are rejected with `RejectedExecutionException`. Today this surfaces as a generic `ExceptionFailure` (which counts toward maxFailures), so repeated shutdown races can exhaust the retry budget and abort the stage even though no real task fault occurred. See SPARK-57465 (reporter: Thomas Newton).
### Design notes
- New dedicated reason mirrors existing special reasons (`ExecutorLostFailure`, `TaskCommitDenied`, `TaskKilled`).
- Gated narrowly on the executor shutdown flag so a genuine non-shutdown `RejectedExecutionException` still counts.
- Resubmission is naturally bounded since the shutting-down executor leaves the cluster.
- `JsonProtocol` ser/deser added. Older event logs without this reason still parse correctly. The forward-compat `MatchError` for older readers of new logs follows the same pre-existing pattern as all `TaskEndReason` additions.
- `TaskSetManager` logs `ExecutorShutdownFailure` at INFO (not WARN) since it is an expected/benign reschedule.
### Does this PR introduce _any_ user-facing change?
No. This is scheduler-internal; the visible effect is fewer spurious stage failures under executor churn.
### How was this patch tested?
- `TaskSetManagerSuite`: the new reason does not increment `numFailures` and the task is resubmitted; contrasted with `ExceptionFailure` which does.
- `JsonProtocolSuite`: round-trip serialization and back-compat deserialization.
### Was this patch authored or co-authored using generative AI tooling?
Tests and PR description generated by Claude Opus 4.8
Closes #56628 from yadavay-amzn/fix/SPARK-57465-rejectedexecution-retries.
Authored-by: Anupam Yadav <an...@amazon.com>
Signed-off-by: Dongjoon Hyun <do...@apache.org>
(cherry picked from commit 7818803c1234559fce80d59b2cbf480fd8555a72)
Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
.../scala/org/apache/spark/TaskEndReason.scala | 12 ++++
.../scala/org/apache/spark/executor/Executor.scala | 10 ++-
.../org/apache/spark/scheduler/DAGScheduler.scala | 2 +-
.../apache/spark/scheduler/TaskSetManager.scala | 6 ++
.../scala/org/apache/spark/util/JsonProtocol.scala | 6 ++
.../org/apache/spark/executor/ExecutorSuite.scala | 84 +++++++++++++++++++++-
.../spark/scheduler/TaskSetManagerSuite.scala | 42 +++++++++++
.../org/apache/spark/util/JsonProtocolSuite.scala | 9 +++
8 files changed, 168 insertions(+), 3 deletions(-)
diff --git a/core/src/main/scala/org/apache/spark/TaskEndReason.scala b/core/src/main/scala/org/apache/spark/TaskEndReason.scala
index f1ce302a05d1..91ed190d7e2f 100644
--- a/core/src/main/scala/org/apache/spark/TaskEndReason.scala
+++ b/core/src/main/scala/org/apache/spark/TaskEndReason.scala
@@ -275,6 +275,18 @@ case class ExecutorLostFailure(
override def countTowardsTaskFailures: Boolean = exitCausedByApp
}
+/**
+ * :: DeveloperApi ::
+ * The task failed because the executor's thread pool rejected it during executor shutdown.
+ * This is not a task fault - the executor was shutting down when the task was submitted.
+ */
+@DeveloperApi
+case class ExecutorShutdownFailure(executorId: String) extends TaskFailedReason {
+ override def toErrorString: String =
+ s"ExecutorShutdownFailure (executor $executorId was shutting down)"
+ override def countTowardsTaskFailures: Boolean = false
+}
+
/**
* :: DeveloperApi ::
* We don't know why the task ended -- for example, because of a ClassNotFound exception when
diff --git a/core/src/main/scala/org/apache/spark/executor/Executor.scala b/core/src/main/scala/org/apache/spark/executor/Executor.scala
index 1790c781ecfc..0de5c376c710 100644
--- a/core/src/main/scala/org/apache/spark/executor/Executor.scala
+++ b/core/src/main/scala/org/apache/spark/executor/Executor.scala
@@ -571,10 +571,18 @@ private[spark] class Executor(
try {
logError(log"Executor launch task ${MDC(TASK_NAME, taskDescription.name)} failed," +
log" reason: ${MDC(REASON, t.getMessage)}")
+ // SPARK-57465: If the thread pool rejected the task because the executor is shutting
+ // down, report a non-counting failure so the task can be rescheduled elsewhere.
+ val reason: TaskFailedReason = t match {
+ case _: RejectedExecutionException if executorShutdown.get() =>
+ ExecutorShutdownFailure(executorId)
+ case _ =>
+ new ExceptionFailure(t, Seq.empty)
+ }
context.statusUpdate(
taskDescription.taskId,
TaskState.FAILED,
- env.closureSerializer.newInstance().serialize(new ExceptionFailure(t, Seq.empty)))
+ env.closureSerializer.newInstance().serialize(reason))
} catch {
case NonFatal(e) if env.isStopped =>
logError(
diff --git a/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala b/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
index 3d615867a927..5e608308de29 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
@@ -2836,7 +2836,7 @@ private[spark] class DAGScheduler(
case TaskResultLost =>
// Do nothing here; the TaskScheduler handles these failures and resubmits the task.
- case _: ExecutorLostFailure | UnknownReason =>
+ case _: ExecutorLostFailure | _: ExecutorShutdownFailure | UnknownReason =>
// Unrecognized failure - also do nothing. If the task fails repeatedly, the TaskScheduler
// will abort the job.
}
diff --git a/core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala b/core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala
index 2e8e6628ea5d..78e4fc78b880 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala
@@ -1087,6 +1087,12 @@ private[spark] class TaskSetManager(
log"Not counting this failure towards the maximum number of failures for the task.")
None
+ case _: ExecutorShutdownFailure =>
+ logInfo(log"${MDC(TASK_NAME, taskName(tid))} failed because its executor was shutting" +
+ log" down while the task was being launched." +
+ log" Not counting this failure towards the maximum number of failures for the task.")
+ None
+
case _: TaskFailedReason => // TaskResultLost and others
logWarning(failureReason)
None
diff --git a/core/src/main/scala/org/apache/spark/util/JsonProtocol.scala b/core/src/main/scala/org/apache/spark/util/JsonProtocol.scala
index d2e684faf9ed..5f7ad0dce1f8 100644
--- a/core/src/main/scala/org/apache/spark/util/JsonProtocol.scala
+++ b/core/src/main/scala/org/apache/spark/util/JsonProtocol.scala
@@ -723,6 +723,8 @@ private[spark] object JsonProtocol extends JsonUtils {
g.writeStringField("Kill Reason", taskKilled.reason)
g.writeFieldName("Accumulator Updates")
accumulablesToJson(taskKilled.accumUpdates, g)
+ case executorShutdownFailure: ExecutorShutdownFailure =>
+ g.writeStringField("Executor ID", executorShutdownFailure.executorId)
case _ =>
// no extra fields to write
}
@@ -1418,6 +1420,7 @@ private[spark] object JsonProtocol extends JsonUtils {
val taskKilled = Utils.getFormattedClassName(TaskKilled)
val taskCommitDenied = Utils.getFormattedClassName(TaskCommitDenied)
val executorLostFailure = Utils.getFormattedClassName(ExecutorLostFailure)
+ val executorShutdownFailure = Utils.getFormattedClassName(ExecutorShutdownFailure)
val unknownReason = Utils.getFormattedClassName(UnknownReason)
}
@@ -1480,6 +1483,9 @@ private[spark] object JsonProtocol extends JsonUtils {
executorId.getOrElse("Unknown"),
exitCausedByApp.getOrElse(true),
reason)
+ case `executorShutdownFailure` =>
+ val executorId = json.get("Executor ID").extractString
+ ExecutorShutdownFailure(executorId)
case `unknownReason` => UnknownReason
}
}
diff --git a/core/src/test/scala/org/apache/spark/executor/ExecutorSuite.scala b/core/src/test/scala/org/apache/spark/executor/ExecutorSuite.scala
index f63cf4f05d61..8b6925b67de7 100644
--- a/core/src/test/scala/org/apache/spark/executor/ExecutorSuite.scala
+++ b/core/src/test/scala/org/apache/spark/executor/ExecutorSuite.scala
@@ -22,7 +22,7 @@ import java.lang.Thread.UncaughtExceptionHandler
import java.net.URL
import java.nio.ByteBuffer
import java.util.{HashMap, Properties}
-import java.util.concurrent.{CountDownLatch, ThreadPoolExecutor, TimeUnit}
+import java.util.concurrent.{CountDownLatch, RejectedExecutionException, ThreadPoolExecutor, TimeUnit}
import java.util.concurrent.atomic.AtomicBoolean
import scala.collection.immutable
@@ -722,6 +722,88 @@ class ExecutorSuite extends SparkFunSuite
}
}
+ test("SPARK-57465: RejectedExecutionException with executorShutdown=true " +
+ "reports ExecutorShutdownFailure") {
+ withRejectedExecutionSetup(executorShutdown = true, "shutting down") { (serializer, captor) =>
+ val failReason = serializer.newInstance()
+ .deserialize[TaskFailedReason](captor.getValue)
+ assert(failReason.isInstanceOf[ExecutorShutdownFailure])
+ assert(failReason.countTowardsTaskFailures === false)
+ assert(failReason.asInstanceOf[ExecutorShutdownFailure].executorId === "id")
+ }
+ }
+
+ test("SPARK-57465: RejectedExecutionException with executorShutdown=false " +
+ "reports ExceptionFailure") {
+ withRejectedExecutionSetup(executorShutdown = false, "pool full") { (serializer, captor) =>
+ val failReason = serializer.newInstance()
+ .deserialize[TaskFailedReason](captor.getValue)
+ assert(failReason.isInstanceOf[ExceptionFailure])
+ assert(failReason.countTowardsTaskFailures === true)
+ val ef = failReason.asInstanceOf[ExceptionFailure]
+ assert(ef.exception.get.isInstanceOf[RejectedExecutionException])
+ }
+ }
+
+ /**
+ * Helper for SPARK-57465 tests: sets up an executor with a mocked threadPool that throws
+ * RejectedExecutionException on execute(), launches a task, captures the statusUpdate,
+ * and yields the serializer + captured ByteBuffer for assertions.
+ */
+ private def withRejectedExecutionSetup(
+ executorShutdown: Boolean,
+ rejectionMessage: String)(
+ f: (JavaSerializer, ArgumentCaptor[ByteBuffer]) => Unit): Unit = {
+ val conf = new SparkConf
+ // Defense-in-depth: set a very large heartbeat interval so the heartbeater thread
+ // cannot fire during this test even if cleanup is delayed.
+ conf.set(EXECUTOR_HEARTBEAT_INTERVAL.key, "3600s")
+ val serializer = new JavaSerializer(conf)
+ val env = createMockEnv(conf, serializer)
+ val serializedTask = serializer.newInstance().serialize(new FakeTask(0, 0))
+ val taskDescription = createFakeTaskDescription(serializedTask)
+
+ val mockExecutorBackend = mock[ExecutorBackend]
+ val statusCaptor = ArgumentCaptor.forClass(classOf[ByteBuffer])
+
+ withExecutor("id", "localhost", env) { executor =>
+ val executorClass = classOf[Executor]
+ val shutdownField = executorClass.getDeclaredField("executorShutdown")
+ shutdownField.setAccessible(true)
+ val shutdownFlag = shutdownField.get(executor).asInstanceOf[AtomicBoolean]
+
+ if (executorShutdown) {
+ shutdownFlag.set(true)
+ }
+
+ val threadPoolField = executorClass.getDeclaredField("threadPool")
+ threadPoolField.setAccessible(true)
+ val mockThreadPool = mock[ThreadPoolExecutor]
+ when(mockThreadPool.execute(any[Runnable]))
+ .thenThrow(new RejectedExecutionException(rejectionMessage))
+ threadPoolField.set(executor, mockThreadPool)
+
+ executor.launchTask(mockExecutorBackend, taskDescription)
+
+ verify(mockExecutorBackend).statusUpdate(
+ meq(taskDescription.taskId),
+ meq(TaskState.FAILED),
+ statusCaptor.capture()
+ )
+
+ f(serializer, statusCaptor)
+
+ // SPARK-57465: Reset executorShutdown to false so that withExecutor's finally block
+ // can call executor.stop() fully. Executor.stop() is guarded by
+ // `if (!executorShutdown.getAndSet(true))` -- leaving the flag true causes stop() to
+ // short-circuit, leaking the heartbeater thread which eventually triggers
+ // System.exit(HEARTBEAT_FAILURE) and crashes the entire test fork.
+ if (executorShutdown) {
+ shutdownFlag.set(false)
+ }
+ }
+ }
+
private def createMockEnv(conf: SparkConf, serializer: JavaSerializer): SparkEnv = {
val mockEnv = mock[SparkEnv]
val mockRpcEnv = mock[RpcEnv]
diff --git a/core/src/test/scala/org/apache/spark/scheduler/TaskSetManagerSuite.scala b/core/src/test/scala/org/apache/spark/scheduler/TaskSetManagerSuite.scala
index 299bce879ba2..28d774eb27f7 100644
--- a/core/src/test/scala/org/apache/spark/scheduler/TaskSetManagerSuite.scala
+++ b/core/src/test/scala/org/apache/spark/scheduler/TaskSetManagerSuite.scala
@@ -3004,6 +3004,48 @@ class TaskSetManagerSuite
new DirectTaskResult[MapStatus](valueSer.serialize(mapStatus), accumUpdates, metricPeaks)
}
+ // SPARK-57465: With the fix, ExecutorShutdownFailure does NOT count toward task failures
+ // (contrast: a normal ExceptionFailure DOES count)
+ test("SPARK-57465: ExecutorShutdownFailure does not count toward task failures") {
+ sc = new SparkContext("local", "test")
+ sched = new FakeTaskScheduler(sc, ("exec1", "host1"))
+ val taskSet = FakeTask.createTaskSet(1)
+ val clock = new ManualClock(1)
+ val manager = new TaskSetManager(sched, taskSet, MAX_TASK_FAILURES, clock = clock)
+
+ val failureReason = ExecutorShutdownFailure("exec1")
+ assert(failureReason.countTowardsTaskFailures === false)
+
+ // Fail the task MAX_TASK_FAILURES times with ExecutorShutdownFailure
+ for (attempt <- 0 until MAX_TASK_FAILURES) {
+ clock.advance(1)
+ val taskOption = manager.resourceOffer("exec1", "host1", ANY)._1
+ assert(taskOption.isDefined, s"Task should still be schedulable at attempt $attempt")
+ manager.handleFailedTask(taskOption.get.taskId, TaskState.FAILED, failureReason)
+ }
+
+ // The stage should NOT be aborted - failures don't count
+ assert(!manager.isZombie,
+ "Stage should NOT be zombie - ExecutorShutdownFailure must not count")
+ assert(!sched.taskSetsFailed.contains("0.0"),
+ "TaskSet should NOT be failed/aborted")
+
+ // Task should still be schedulable
+ clock.advance(1)
+ val taskOption = manager.resourceOffer("exec1", "host1", ANY)._1
+ assert(taskOption.isDefined, "Task should still be schedulable after non-counting failures")
+
+ // Contrast: a normal ExceptionFailure DOES count toward task failures
+ val exceptionFailure = new ExceptionFailure(
+ "java.lang.RuntimeException", "test failure",
+ Array.empty, "java.lang.RuntimeException: test failure", None)
+ assert(exceptionFailure.countTowardsTaskFailures === true)
+ manager.handleFailedTask(taskOption.get.taskId, TaskState.FAILED, exceptionFailure)
+ val numFailures = PrivateMethod[Array[Int]](Symbol("numFailures"))
+ assert(manager.invokePrivate(numFailures())(0) === 1,
+ "ExceptionFailure must count towards task failures (contrast)")
+ }
+
}
class FakeLongTasks(stageId: Int, partitionId: Int) extends FakeTask(stageId, partitionId) {
diff --git a/core/src/test/scala/org/apache/spark/util/JsonProtocolSuite.scala b/core/src/test/scala/org/apache/spark/util/JsonProtocolSuite.scala
index 70edf2b7db43..0dbd8d792137 100644
--- a/core/src/test/scala/org/apache/spark/util/JsonProtocolSuite.scala
+++ b/core/src/test/scala/org/apache/spark/util/JsonProtocolSuite.scala
@@ -254,6 +254,7 @@ class JsonProtocolSuite extends SparkFunSuite {
testTaskEndReason(TaskKilled("test"))
testTaskEndReason(TaskCommitDenied(2, 3, 4))
testTaskEndReason(ExecutorLostFailure("100", true, Some("Induced failure")))
+ testTaskEndReason(ExecutorShutdownFailure("exec-42"))
testTaskEndReason(UnknownReason)
// BlockId
@@ -478,6 +479,12 @@ class JsonProtocolSuite extends SparkFunSuite {
assert(expectedExecutorLostFailure === JsonProtocol.taskEndReasonFromJson(oldEvent))
}
+ test("SPARK-57465: ExecutorShutdownFailure deserializes from its JSON representation") {
+ val json = """{"Reason":"ExecutorShutdownFailure","Executor ID":"exec-42"}"""
+ val reason = JsonProtocol.taskEndReasonFromJson(json)
+ assert(reason === ExecutorShutdownFailure("exec-42"))
+ }
+
test("SparkListenerJobStart backward compatibility") {
// Prior to Spark 1.2.0, SparkListenerJobStart did not have a "Stage Infos" property.
val stageIds = Seq[Int](1, 2, 3, 4)
@@ -1359,6 +1366,8 @@ private[spark] object JsonProtocolSuite extends Assertions {
assert(execId1 === execId2)
assert(exit1CausedByApp === exit2CausedByApp)
assert(reason1 === reason2)
+ case (ExecutorShutdownFailure(execId1), ExecutorShutdownFailure(execId2)) =>
+ assert(execId1 === execId2)
case (UnknownReason, UnknownReason) =>
case _ => fail("Task end reasons don't match in types!")
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org