gaoyajun02 commented on code in PR #58008:
URL: https://github.com/apache/spark/pull/58008#discussion_r3851634243
##########
core/src/main/scala/org/apache/spark/internal/config/package.scala:
##########
@@ -2973,6 +2973,27 @@ package object config {
.doubleConf
.createWithDefault(1.0)
+ private[spark] val STALE_PUSH_FALLBACK_ENABLED =
+ ConfigBuilder("spark.shuffle.push.stale.fallback.enabled")
+ .doc("When true, mark partitions with stale push data so reducers
fallback to " +
+ "unmerged blocks. Default false to keep detection observational only:
duplicate map " +
+ "attempts are always reported (logged), but reducer fallback must be
explicitly " +
+ "enabled. When enabled, fallback marking applies to indeterminate
stages only unless " +
+ "spark.shuffle.push.stale.detectAllStages.enabled is also true.")
+ .version("4.4.0")
+ .booleanConf
+ .createWithDefault(true)
Review Comment:
Done. Changed `createWithDefault(true)` -> `createWithDefault(false)` so the
registered default matches the "Default false" documentation and the opt-in
contract. The default no longer re-enables reducer fallback for indeterminate
stages without an explicit setting.
Verified the only reader is `TaskSetManager.stalePushFallbackEnabled`; tests
set the config explicitly, so no test relied on the `true` default.
##########
core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala:
##########
@@ -82,6 +82,19 @@ private[spark] class TaskSetManager(
private val isShuffleMapTasks = tasks(0).isInstanceOf[ShuffleMapTask]
// shuffleId is only available when isShuffleMapTasks=true
private val shuffleId = taskSet.shuffleId
+ // Scopes stale-push reducer fallback to indeterminate stages (deterministic
stages reproduce
+ // identical output across attempts, so a stale push there is benign).
Defaults to false when the
+ // stage isn't registered (e.g. tests). Uses taskSet.stageId (constructor
param) because the
+ // `stageId` field below is not initialized yet at this source position.
+ private val isIndeterminateShuffleMapStage: Boolean = isShuffleMapTasks &&
Review Comment:
Done. `TaskSetManager` no longer caches indeterminacy as a `Boolean` at
construction. It now holds the `Option[ShuffleMapStage]` reference and
re-evaluates `isStaticallyIndeterminate || isRuntimeIndeterminate` inside
`detectStalePushIfShuffleTask`, when the late duplicate result arrives. A
checksum mismatch set by `DAGScheduler` during the stage (via
`registerMapOutput`) is therefore observed rather than read from a stale
constructor snapshot.
Added a test case (`fallbackEnabled=true, detectAllStages=false`) that sets
`isChecksumMismatched` after the `TaskSetManager` is constructed and asserts
stale marking still fires — this would fail under the old cached-`Boolean`
approach.
--
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]
