You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "jiangxb1987 (via GitHub)" <gi...@apache.org> on 2026/04/15 22:48:20 UTC
[PR] [SPARK-56496][CORE] Unify stage retry limit checks into canRetryStage helper in DAGScheduler [spark]
jiangxb1987 opened a new pull request, #55360:
URL: https://github.com/apache/spark/pull/55360
### What changes were proposed in this pull request?
- Introduce a `canRetryStage(stage)` private helper in `DAGScheduler` that enforces both retry limits in one place: `maxConsecutiveStageAttempts` (caps consecutive failures, cleared on success) and `maxStageAttempts` (absolute total-attempts ceiling, never cleared).
- Replace three independent inline `shouldAbortStage` computations (non-fetch failure path, fetch-failure path, barrier-stage failure path) — each of which previously only checked the consecutive limit — with calls to the new helper.
- Both limits are now documented as *independently enforced*; whichever triggers first aborts the stage.
- Abort messages now correctly identify which limit was exceeded (consecutive vs. total).
- Add unit test that verifies the total-attempts limit is enforced in the failure path even when the consecutive counter was cleared by an intervening stage success.
### Why are the changes needed?
Fix the inconsistent behavior between fetch-failure paths and non-fetch-failure paths.
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
Verified by a new unit test in DAGSchedulerSuite.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Opus 4.6
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56496][CORE] Unify stage retry limit checks into canRetryStage helper in DAGScheduler [spark]
Posted by "jiangxb1987 (via GitHub)" <gi...@apache.org>.
jiangxb1987 commented on code in PR #55360:
URL: https://github.com/apache/spark/pull/55360#discussion_r3103258423
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -2131,17 +2156,19 @@ private[spark] class DAGScheduler(
}
stage.failedAttemptIds.add(stage.latestInfo.attemptNumber())
- val shouldAbortStage = stage.failedAttemptIds.size >= maxConsecutiveStageAttempts ||
- disallowStageRetryForTest
+ val shouldAbortStage = !canRetryStage(stage)
markStageAsFinished(stage, Some(reason), willRetry = !shouldAbortStage)
if (shouldAbortStage) {
val abortMessage = if (disallowStageRetryForTest) {
"Stage will not retry stage due to testing config. Most recent failure " +
s"reason: $reason"
- } else {
+ } else if (stage.failedAttemptIds.size >= maxConsecutiveStageAttempts) {
s"$stage (${stage.name}) has failed the maximum allowable number of " +
s"times: $maxConsecutiveStageAttempts. Most recent failure reason: $reason"
+ } else {
+ s"$stage (${stage.name}) has exceeded the maximum allowable total stage " +
+ s"attempts: $maxStageAttempts. Most recent failure reason: $reason"
}
Review Comment:
Updated. Replaced `canRetryStage(Boolean)` with `stageAbortReason(stage, failureMessage): Option[String]` that returns the abort message describing which limit was hit. Each call site now just uses `abortReason.get` instead of re-evaluating the conditions.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56496][CORE] Unify stage retry limit checks into canRetryStage helper in DAGScheduler [spark]
Posted by "jiangxb1987 (via GitHub)" <gi...@apache.org>.
jiangxb1987 commented on code in PR #55360:
URL: https://github.com/apache/spark/pull/55360#discussion_r3103259865
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -247,17 +247,42 @@ private[spark] class DAGScheduler(
/**
* Number of consecutive stage attempts allowed before a stage is aborted.
+ * The consecutive counter is cleared when a stage succeeds, so failures from different
+ * "runs" of the same stage do not accumulate against this limit.
+ * See [[canRetryStage]] for how this interacts with [[maxStageAttempts]].
*/
private[scheduler] val maxConsecutiveStageAttempts =
sc.conf.get(config.STAGE_MAX_CONSECUTIVE_ATTEMPTS)
/**
- * Max stage attempts allowed before a stage is aborted.
+ * Absolute ceiling on the total number of attempts ever made for a stage (consecutive or not).
+ * Computed as the maximum of [[maxConsecutiveStageAttempts]] and the configured value of
+ * [[config.STAGE_MAX_ATTEMPTS]], so that the total ceiling is never lower than the consecutive
+ * limit. See [[canRetryStage]] for how this interacts with [[maxConsecutiveStageAttempts]].
*/
private[scheduler] val maxStageAttempts: Int = {
Math.max(maxConsecutiveStageAttempts, sc.conf.get(config.STAGE_MAX_ATTEMPTS))
}
+ /**
+ * Returns true if the stage may be retried after a failure, false if it should be aborted.
+ *
+ * Two independent limits are enforced; whichever triggers first causes the stage to be aborted:
+ * - [[maxConsecutiveStageAttempts]]: caps the number of *consecutive* failed attempts.
+ * This counter is reset when the stage succeeds, so transient failures in a long-running
+ * job do not permanently accumulate.
+ * - [[maxStageAttempts]]: an absolute ceiling on the *total* number of attempts ever made
+ * for a stage. This counter is never cleared.
Review Comment:
Fixed, thanks!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56496][CORE] Unify stage retry limit checks into canRetryStage helper in DAGScheduler [spark]
Posted by "asf-gitbox-commits (via GitHub)" <gi...@apache.org>.
asf-gitbox-commits closed pull request #55360: [SPARK-56496][CORE] Unify stage retry limit checks into canRetryStage helper in DAGScheduler
URL: https://github.com/apache/spark/pull/55360
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56496][CORE] Unify stage retry limit checks into canRetryStage helper in DAGScheduler [spark]
Posted by "Ngone51 (via GitHub)" <gi...@apache.org>.
Ngone51 commented on code in PR #55360:
URL: https://github.com/apache/spark/pull/55360#discussion_r3090597164
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -247,17 +247,41 @@ private[spark] class DAGScheduler(
/**
* Number of consecutive stage attempts allowed before a stage is aborted.
+ * The consecutive counter is cleared when a stage succeeds, so failures from different
+ * "runs" of the same stage do not accumulate against this limit.
+ * See [[canRetryStage]] for how this interacts with [[maxStageAttempts]].
*/
private[scheduler] val maxConsecutiveStageAttempts =
sc.conf.get(config.STAGE_MAX_CONSECUTIVE_ATTEMPTS)
/**
- * Max stage attempts allowed before a stage is aborted.
+ * Absolute ceiling on the total number of attempts ever made for a stage (consecutive or not).
+ * Computed as the maximum of [[maxConsecutiveStageAttempts]] and the configured value of
+ * [[config.STAGE_MAX_ATTEMPTS]], so that the total ceiling is never lower than the consecutive
+ * limit. See [[canRetryStage]] for how this interacts with [[maxConsecutiveStageAttempts]].
*/
private[scheduler] val maxStageAttempts: Int = {
Math.max(maxConsecutiveStageAttempts, sc.conf.get(config.STAGE_MAX_ATTEMPTS))
}
+ /**
+ * Returns true if the stage may be retried after a failure, false if it should be aborted.
+ *
+ * Two independent limits are enforced; whichever triggers first causes the stage to be aborted:
+ * - [[maxConsecutiveStageAttempts]]: caps the number of *consecutive* failed attempts.
+ * This counter is reset when the stage succeeds, so transient failures in a long-running
+ * job do not permanently accumulate.
+ * - [[maxStageAttempts]]: an absolute ceiling on the *total* number of attempts ever made
+ * for a stage. This counter is never cleared.
+ *
+ * Note: the test-only [[disallowStageRetryForTest]] flag forces an abort regardless of these
Review Comment:
Why don't we also incorporate `disallowStageRetryForTest` in `canRetryStage()`?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56496][CORE] Unify stage retry limit checks into canRetryStage helper in DAGScheduler [spark]
Posted by "jiangxb1987 (via GitHub)" <gi...@apache.org>.
jiangxb1987 commented on code in PR #55360:
URL: https://github.com/apache/spark/pull/55360#discussion_r3095434491
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -247,17 +247,41 @@ private[spark] class DAGScheduler(
/**
* Number of consecutive stage attempts allowed before a stage is aborted.
+ * The consecutive counter is cleared when a stage succeeds, so failures from different
+ * "runs" of the same stage do not accumulate against this limit.
+ * See [[canRetryStage]] for how this interacts with [[maxStageAttempts]].
*/
private[scheduler] val maxConsecutiveStageAttempts =
sc.conf.get(config.STAGE_MAX_CONSECUTIVE_ATTEMPTS)
/**
- * Max stage attempts allowed before a stage is aborted.
+ * Absolute ceiling on the total number of attempts ever made for a stage (consecutive or not).
+ * Computed as the maximum of [[maxConsecutiveStageAttempts]] and the configured value of
+ * [[config.STAGE_MAX_ATTEMPTS]], so that the total ceiling is never lower than the consecutive
+ * limit. See [[canRetryStage]] for how this interacts with [[maxConsecutiveStageAttempts]].
*/
private[scheduler] val maxStageAttempts: Int = {
Math.max(maxConsecutiveStageAttempts, sc.conf.get(config.STAGE_MAX_ATTEMPTS))
}
+ /**
+ * Returns true if the stage may be retried after a failure, false if it should be aborted.
+ *
+ * Two independent limits are enforced; whichever triggers first causes the stage to be aborted:
+ * - [[maxConsecutiveStageAttempts]]: caps the number of *consecutive* failed attempts.
+ * This counter is reset when the stage succeeds, so transient failures in a long-running
+ * job do not permanently accumulate.
+ * - [[maxStageAttempts]]: an absolute ceiling on the *total* number of attempts ever made
+ * for a stage. This counter is never cleared.
+ *
+ * Note: the test-only [[disallowStageRetryForTest]] flag forces an abort regardless of these
Review Comment:
good catch! updated.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56496][CORE] Unify stage retry limit checks into canRetryStage helper in DAGScheduler [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #55360:
URL: https://github.com/apache/spark/pull/55360#discussion_r3097872768
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -2131,17 +2156,19 @@ private[spark] class DAGScheduler(
}
stage.failedAttemptIds.add(stage.latestInfo.attemptNumber())
- val shouldAbortStage = stage.failedAttemptIds.size >= maxConsecutiveStageAttempts ||
- disallowStageRetryForTest
+ val shouldAbortStage = !canRetryStage(stage)
markStageAsFinished(stage, Some(reason), willRetry = !shouldAbortStage)
if (shouldAbortStage) {
val abortMessage = if (disallowStageRetryForTest) {
"Stage will not retry stage due to testing config. Most recent failure " +
s"reason: $reason"
- } else {
+ } else if (stage.failedAttemptIds.size >= maxConsecutiveStageAttempts) {
s"$stage (${stage.name}) has failed the maximum allowable number of " +
s"times: $maxConsecutiveStageAttempts. Most recent failure reason: $reason"
+ } else {
+ s"$stage (${stage.name}) has exceeded the maximum allowable total stage " +
+ s"attempts: $maxStageAttempts. Most recent failure reason: $reason"
}
Review Comment:
This 3-way abort-message branch is now duplicated three times (here, ~2474, ~2609). `canRetryStage` consolidated the decision but this companion logic is still triplicated — and each copy re-evaluates `failedAttemptIds.size >= maxConsecutiveStageAttempts`, which `canRetryStage` already computed internally.
Consider extending the consolidation: have `canRetryStage` (or a sibling helper) return `Option[String]` / a small ADT indicating which limit tripped, and let each call site just substitute the context-specific failure-reason suffix. That would collapse each of the three blocks to ~3 lines and remove the redundant check. Optional — matches the PR's stated goal of "enforces both retry limits in one place."
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -247,17 +247,42 @@ private[spark] class DAGScheduler(
/**
* Number of consecutive stage attempts allowed before a stage is aborted.
+ * The consecutive counter is cleared when a stage succeeds, so failures from different
+ * "runs" of the same stage do not accumulate against this limit.
+ * See [[canRetryStage]] for how this interacts with [[maxStageAttempts]].
*/
private[scheduler] val maxConsecutiveStageAttempts =
sc.conf.get(config.STAGE_MAX_CONSECUTIVE_ATTEMPTS)
/**
- * Max stage attempts allowed before a stage is aborted.
+ * Absolute ceiling on the total number of attempts ever made for a stage (consecutive or not).
+ * Computed as the maximum of [[maxConsecutiveStageAttempts]] and the configured value of
+ * [[config.STAGE_MAX_ATTEMPTS]], so that the total ceiling is never lower than the consecutive
+ * limit. See [[canRetryStage]] for how this interacts with [[maxConsecutiveStageAttempts]].
*/
private[scheduler] val maxStageAttempts: Int = {
Math.max(maxConsecutiveStageAttempts, sc.conf.get(config.STAGE_MAX_ATTEMPTS))
}
+ /**
+ * Returns true if the stage may be retried after a failure, false if it should be aborted.
+ *
+ * Two independent limits are enforced; whichever triggers first causes the stage to be aborted:
+ * - [[maxConsecutiveStageAttempts]]: caps the number of *consecutive* failed attempts.
+ * This counter is reset when the stage succeeds, so transient failures in a long-running
+ * job do not permanently accumulate.
+ * - [[maxStageAttempts]]: an absolute ceiling on the *total* number of attempts ever made
+ * for a stage. This counter is never cleared.
Review Comment:
Minor scaladoc nit: "This counter is never cleared." Grammatically "this counter" reads as pointing back to `maxStageAttempts`, which is a *limit*, not a counter. The actual never-cleared counter is `nextAttemptId`.
```suggestion
* - [[maxStageAttempts]]: an absolute ceiling on the *total* number of attempts ever made
* for a stage. The attempt counter it bounds (`nextAttemptId`) is never cleared.
```
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56496][CORE] Unify stage retry limit checks into canRetryStage helper in DAGScheduler [spark]
Posted by "jiangxb1987 (via GitHub)" <gi...@apache.org>.
jiangxb1987 commented on PR #55360:
URL: https://github.com/apache/spark/pull/55360#issuecomment-4290983981
Thanks, merging to master!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org