[SPARK-58185][CORE] Define PipelinedShuffleDependency and route shuffles to a ShuffleManager by dependency type

### What changes were proposed in this pull request?

This PR routes each shuffle to a `ShuffleManager` **by its dependency type**, so a
`PipelinedShuffleDependency` (added in the previous PR) is served by an incremental shuffle
implementation while every other shuffle continues to use the regular one.

`SparkEnv` now holds two shuffle managers, keyed by **kind** rather than a default and an override:

- `spark.shuffle.manager` — the **blocking** manager (a `BlockingShuffleManager`), serving all
  regular (materialized) shuffles and owning block-by-id resolution. Defaults to `sort`, as before.
- `spark.shuffle.manager.incremental` — the **pipelined** manager (a `PipelinedShuffleManager`),
  serving pipelined (incrementally-readable) shuffle dependencies. Defaults to the built-in
  `streaming` manager, just as the blocking manager defaults to `sort`.

Routing is a single pure function, `SparkEnv.shuffleManagerFor(dependency)`: a
`PipelinedShuffleDependency` goes to the pipelined manager, every other `ShuffleDependency` to the
blocking manager. Both managers always exist, so there is no fallback path. Because the decision is
a pure function of the (serialized, deterministic) dependency, the driver at `registerShuffle` and
every executor at `getWriter` / `getReader` agree without any shared routing state or config
re-read. All shuffle-I/O sites route through it (`Dependency.shuffleHandle`, `ShuffleWriteProcessor`,
`ShuffledRDD`, `CoGroupedRDD`, `SubtractedRDD`, `ShuffledRowRDD`, `ShuffleExchangeExec`).

To make the manager APIs harder to misuse, `ShuffleManager` is split by kind:

```
ShuffleManager                 (declare a kind by extending one of the two subtypes)
  BlockingShuffleManager       <- materialized output, served as block-manager blocks via a
                                   ShuffleBlockResolver (SortShuffleManager)
  PipelinedShuffleManager      <- incrementally-readable, served out-of-band, no resolver
                                   (StreamingShuffleManager)
```

`shuffleBlockResolver` moves off `ShuffleManager` onto `BlockingShuffleManager`, so a manager's type
now says whether it serves block-manager-addressed blocks at all -- instead of the old contract
where every manager had a `shuffleBlockResolver` that some implementations threw from. Block-by-id
resolution (reads, push-based merge, decommission migration, external-shuffle-service cleanup) is
served through a new `SparkEnv.shuffleBlockResolver: Option[ShuffleBlockResolver]`, sourced from the
blocking manager; it is `None` only before the shuffle manager is initialized. This is correct
because a pipelined shuffle is served out-of-band and produces no block-manager blocks, so those
paths only ever resolve regular shuffles.

`initializeShuffleManager` validates each slot's kind and fails fast otherwise: the
`spark.shuffle.manager` slot must be a `BlockingShuffleManager` and the
`spark.shuffle.manager.incremental` slot a `PipelinedShuffleManager`. A manager that declares
neither kind is rejected from both slots. This is the "declare your kind" enforcement, and it
applies to reflectively-loaded managers too, so `ShuffleManager` itself is **not** sealed -- the
SPARK-45762 third-party extension point (a user-jar manager implementing the interface) stays open.

Supporting changes:

- `SparkEnv.blockingShuffleManager` and `SparkEnv.pipelinedShuffleManager` are the explicit
  by-kind accessors (e.g. sort-shuffle detection in `ShuffleExchangeExec` reads
  `blockingShuffleManager`); the bare `SparkEnv.shuffleManager` accessor is deprecated in favor of
  `shuffleManagerFor` / `blockingShuffleManager` so the routing decision is explicit at each call
  site.
- `spark.shuffle.manager.incremental` accepts the same style of short aliases as
  `spark.shuffle.manager` -- `streaming` (the default) resolves to `StreamingShuffleManager`, and
  `sort` / `tungsten-sort` resolve to `SortShuffleManager` (though a blocking manager is then
  rejected from this slot by the kind validation).
- The `StreamingShuffleOutputTracker` is initialized when the pipelined manager is a
  `StreamingShuffleManager` (the default) or the blocking manager is a `MultiShuffleManager`.
- `MultiShuffleManager` (the legacy cluster-level single-slot way to mix streaming and regular
  shuffle) is deprecated in favor of this per-dependency routing. The routing code does not use it;
  `SparkEnv` still recognizes it for tracker initialization (with a TODO to drop that once
  `MultiShuffleManager` is removed).
- `spark.shuffle.manager.incremental` declares `ConfigBindingPolicy.NOT_APPLICABLE` (a shuffle
  manager is a physical-execution choice and does not change how a view/UDF body resolves).

This is a follow-up in the stack that begins with `PipelinedShuffleDependency`; a later
`DAGScheduler` change adds the concurrent (pipelined-group) stage scheduling that actually
constructs these dependencies.

### Why are the changes needed?

`PipelinedShuffleDependency` marks a shuffle whose consumer reads output incrementally, but a marker
type alone does nothing until the shuffle layer can serve it with an incremental implementation
while regular shuffles keep using the existing one. A cluster needs both to coexist and to be chosen
per shuffle, not per cluster. This PR provides that: a dependency-typed routing point plus two
coexisting managers keyed by kind, so batch (sort) and pipelined (e.g. streaming real-time mode)
shuffles can run in the same application, each served by the right implementation. Splitting
`ShuffleManager` by kind (`BlockingShuffleManager` / `PipelinedShuffleManager`) encodes in the type
system the invariant the routing relies on -- that only a blocking manager serves
block-manager-addressed blocks -- so callers can no longer accidentally reach a resolver that does
not exist.

### Does this PR introduce _any_ user-facing change?

No user-visible behavior change for existing workloads: regular shuffles are still served by
`spark.shuffle.manager` (default `sort`) exactly as before, query results are unchanged, and the
block-by-id paths (reads, push-based merge, decommission, ESS cleanup) are unchanged. Nothing
constructs a `PipelinedShuffleDependency` yet, so no shuffle is routed to the pipelined manager in
this PR.

One internal behavior change worth calling out: `spark.shuffle.manager.incremental` now defaults to
the built-in streaming manager (rather than being unset), so a `StreamingShuffleOutputTracker` is
initialized in every `SparkEnv`. The new config, the `BlockingShuffleManager` /
`PipelinedShuffleManager` traits, and the `SparkEnv` routing accessors are `private[spark]`; the only
externally visible API change is that the `DeveloperApi` `SparkEnv.shuffleManager` accessor is now
`deprecated` (still functional, pointing callers to the explicit accessors).

### How was this patch tested?

New unit tests in `PipelinedShuffleRoutingSuite` (16 tests) verify the routing and the manager-kind
model:

- `shuffleManagerFor` routes a regular dependency to the blocking manager and a
  `PipelinedShuffleDependency` to the pipelined manager;
- the handle a dependency mints comes from the routed manager (so driver and executor agree), and
  `blockingShuffleManager` is the plain configured manager (no wrapper installed in front of it);
- the id-only cleanup path notifies both managers (and is a no-op, not an NPE, before the managers
  are initialized), and both managers are stopped on `SparkContext` stop;
- `spark.shuffle.manager.incremental` resolves the same short aliases as `spark.shuffle.manager`, a
  bad incremental class name fails fast at startup, a blocking manager in the incremental slot is
  rejected, and a pipelined manager in the default slot is rejected;
- a manager's type declares its kind (`SortShuffleManager` is a `BlockingShuffleManager`;
  `StreamingShuffleManager` is a `PipelinedShuffleManager` and not a `BlockingShuffleManager`), and
  `SparkEnv.shuffleBlockResolver` is defined for the blocking manager.

`StreamingShuffleManagerSuite` verifies the streaming output tracker is initialized by default (the
streaming incremental manager), for an explicit incremental `StreamingShuffleManager`, and for a
`MultiShuffleManager` default. Existing coverage of the block-by-id and cleanup paths was updated
and rerun: `BlockManagerSuite` (including the deferred-`ShuffleManager` init and unsupported-resolver
paths), `ContextCleanerSuite` (the real `RemoveShuffle` RPC through `BlockManagerMasterEndpoint`),
`SortShuffleSuite`, `SparkSubmitSuite` (the SPARK-45762 user-jar `ShuffleManager` plugin, now
implementing `BlockingShuffleManager`), and `ShuffleDependencySuite` (which also asserts a
`PipelinedShuffleDependency` never allows push-based shuffle merge).

```
build/sbt 'core/testOnly org.apache.spark.shuffle.PipelinedShuffleRoutingSuite'
...
Tests: succeeded 16, failed 0
```

### Was this patch authored or co-authored using generative AI tooling?

Co-authored: Claude Code (Opus 4.8)

Closes #57286 from jerrypeng/stack/pipelined-shuffle-pr2-routing.

Authored-by: Boyang Jerry Peng <jerry.peng@databricks.com>
Signed-off-by: Wenchen Fan <wenchen@databricks.com>

Author: jerrypeng

core/src/main/scala/org/apache/spark/Dependency.scala

@@ -134,7 +134,10 @@ class ShuffleDependency[K: ClassTag, V: ClassTag, C: ClassTag](
   // is enabled
   private[this] var _shuffleMergeAllowed = canShuffleMergeBeEnabled()
 
-  val shuffleHandle: ShuffleHandle = _rdd.context.env.shuffleManager.registerShuffle(
+  // Route to the manager for this dependency's type (the incremental manager for a
+  // PipelinedShuffleDependency, the default manager otherwise). The handle is minted once here on
+  // the driver, so executors that later read it are served by the same manager.
+  val shuffleHandle: ShuffleHandle = _rdd.context.env.shuffleManagerFor(this).registerShuffle(
     shuffleId, this)
 
   private[spark] def setShuffleMergeAllowed(shuffleMergeAllowed: Boolean): Unit = {
@@ -258,6 +261,74 @@ class ShuffleDependency[K: ClassTag, V: ClassTag, C: ClassTag](
 }
 
 
+/**
+ * :: DeveloperApi ::
+ * A [[ShuffleDependency]] whose output can be read incrementally: a consumer stage may begin
+ * reading the shuffle output while the producer stage is still running, rather than waiting for the
+ * producer's full, materialized output.
+ *
+ * This is a subtype of [[ShuffleDependency]] -- and thus, like it, a first-class dependency kind
+ * alongside [[NarrowDependency]] under [[Dependency]]. It is intended to be the marker the
+ * `DAGScheduler` will use to decide that the producer and consumer
+ * stages connected by this edge may run concurrently (a "pipelined group"), and that the shuffle
+ * layer should serve this shuffle with an incremental shuffle implementation. A plain
+ * [[ShuffleDependency]] keeps the existing semantics: its output is fully materialized before any
+ * consumer reads it.
+ *
+ * Two behaviors are active from construction, driven by this type: shuffle registration routes to
+ * the incremental (pipelined) shuffle manager via `SparkEnv.shuffleManagerFor` (a plain
+ * [[ShuffleDependency]] goes to the blocking manager), and push-based shuffle merge is
+ * unconditionally disabled (`setShuffleMergeAllowed(false)`; see below) because merge exposes
+ * output only after a post-completion finalize step and would register merge results for a
+ * transient shuffle. Beyond those, the concurrent scheduling behavior is added separately by the
+ * `DAGScheduler` components that match on this type; code that only matches the parent
+ * `ShuffleDependency` still treats it as an ordinary (materialized) shuffle for everything else.
+ *
+ * The name is *pipelined* rather than *streaming*: reading producer output as it is produced is a
+ * general execution capability (software-pipelining of dependent stages), not specific to
+ * streaming. Streaming / real-time mode is the first caller, but nothing here is
+ * streaming-specific.
+ *
+ * Note: the parent's `checksumMismatchFullRetryEnabled` /
+ * `checksumMismatchQueryLevelRollbackEnabled` parameters are intentionally not exposed here, so
+ * they stay at their `false` defaults for a pipelined shuffle. Their checksum retry / query-level
+ * rollback recomputes and re-runs succeeding stages after a mismatch; in a pipelined group any
+ * failure aborts the whole group and the caller reruns from scratch, so that stage-level recompute
+ * never fires -- the mechanism is moot by construction (it is also incompatible with a consumer
+ * that has already read the output incrementally). Leaving the params unset keeps the idiom
+ * unreachable.
+ */
+@DeveloperApi
+class PipelinedShuffleDependency[K: ClassTag, V: ClassTag, C: ClassTag](
+    _rdd: RDD[_ <: Product2[K, V]],
+    partitioner: Partitioner,
+    serializer: Serializer = SparkEnv.get.serializer,
+    keyOrdering: Option[Ordering[K]] = None,
+    aggregator: Option[Aggregator[K, V, C]] = None,
+    mapSideCombine: Boolean = false,
+    shuffleWriterProcessor: ShuffleWriteProcessor = new ShuffleWriteProcessor,
+    rowBasedChecksums: Array[RowBasedChecksum] = ShuffleDependency.EMPTY_ROW_BASED_CHECKSUMS)
+  extends ShuffleDependency[K, V, C](
+    _rdd,
+    partitioner,
+    serializer,
+    keyOrdering,
+    aggregator,
+    mapSideCombine,
+    shuffleWriterProcessor,
+    rowBasedChecksums) {
+
+  // Push-based shuffle merge is incompatible with a pipelined (incrementally-readable) shuffle: it
+  // exposes output only after a post-completion "finalize" step, the opposite of incremental reads,
+  // and would register merge results in the MapOutputTracker for a transient shuffle that must not
+  // outlive its group. ShuffleDependency enables merge by default whenever push-based shuffle is on
+  // cluster-wide; disable it here so a pipelined producer never acquires merger locations. Without
+  // this, ShuffleWriteProcessor.write would reach the push path and dereference the incremental
+  // manager's shuffleBlockResolver, which the streaming manager does not support.
+  setShuffleMergeAllowed(false)
+}
+
+
 /**
  * :: DeveloperApi ::
  * Represents a one-to-one dependency between partitions of the parent and child RDDs.

core/src/main/scala/org/apache/spark/SparkEnv.scala

@@ -45,7 +45,8 @@ import org.apache.spark.scheduler.{LiveListenerBus, OutputCommitCoordinator}
 import org.apache.spark.scheduler.OutputCommitCoordinator.OutputCommitCoordinatorEndpoint
 import org.apache.spark.security.CryptoStreamUtils
 import org.apache.spark.serializer.{JavaSerializer, Serializer, SerializerManager}
-import org.apache.spark.shuffle.ShuffleManager
+import org.apache.spark.shuffle.{BlockingShuffleManager, PipelinedShuffleManager}
+import org.apache.spark.shuffle.{ShuffleBlockResolver, ShuffleManager}
 import org.apache.spark.shuffle.streaming.{MultiShuffleManager, StreamingShuffleManager}
 import org.apache.spark.storage._
 import org.apache.spark.udf.worker.UDFWorkerSpecification
@@ -76,15 +77,105 @@ class SparkEnv (
     val outputCommitCoordinator: OutputCommitCoordinator,
     val conf: SparkConf) extends Logging {
 
-  // We initialize the ShuffleManager later in SparkContext and Executor to allow
-  // user jars to define custom ShuffleManagers.
-  @volatile private var _shuffleManager: ShuffleManager = _
+  // The two shuffle managers are peers keyed by kind, not a default and an override: a shuffle is
+  // routed to one or the other by its dependency type via `shuffleManagerFor`, so neither is ever
+  // installed "behind" the other.
+  //
+  // The blocking manager (spark.shuffle.manager) serves all regular, materialized shuffles and owns
+  // block-by-id resolution. It is always a BlockingShuffleManager -- initializeShuffleManager
+  // rejects a non-blocking manager in this slot -- so block-resolution has a single, well-typed
+  // source. We initialize it later in SparkContext and Executor to allow user jars to define custom
+  // ShuffleManagers.
+  @volatile private var _blockingShuffleManager: BlockingShuffleManager = _
+
+  // The pipelined manager (spark.shuffle.manager.incremental) serves pipelined
+  // (incrementally-readable) shuffle dependencies. It defaults to the built-in
+  // StreamingShuffleManager -- just as the blocking manager defaults to sort -- and is always a
+  // PipelinedShuffleManager (initializeShuffleManager rejects a blocking manager in this slot), so
+  // its output is never expected to be reachable through the block-manager resolver. Like the
+  // blocking manager, it is initialized later in SparkContext and Executor to allow user jars.
+  @volatile private var _pipelinedShuffleManager: PipelinedShuffleManager = _
 
   // Latch to signal when the ShuffleManager has been initialized.
   // Used to allow callers to wait for initialization.
   private val shuffleManagerInitLatch = new CountDownLatch(1)
 
-  def shuffleManager: ShuffleManager = _shuffleManager
+  /**
+   * The `BlockingShuffleManager` (configured by spark.shuffle.manager), which serves all regular,
+   * materialized shuffle dependencies and owns block-by-id resolution. Use this when the intent is
+   * specifically the blocking manager -- e.g. inspecting its concrete type. To serve a specific
+   * shuffle's reads/writes, use `shuffleManagerFor`, which routes by dependency type; to resolve a
+   * block by id, use `shuffleBlockResolver`.
+   */
+  private[spark] def blockingShuffleManager: BlockingShuffleManager = _blockingShuffleManager
+
+  /**
+   * The `PipelinedShuffleManager` (configured by spark.shuffle.manager.incremental, defaulting to
+   * the built-in streaming manager) that serves pipelined shuffle dependencies. A pipelined shuffle
+   * is read incrementally and served out-of-band, so this manager never provides a
+   * `ShuffleBlockResolver` of its own.
+   */
+  private[spark] def pipelinedShuffleManager: PipelinedShuffleManager = _pipelinedShuffleManager
+
+  /**
+   * The `ShuffleBlockResolver` used to resolve shuffle blocks by id (reads, push-merge, and
+   * decommission migration), or `None` before the shuffle manager is initialized. Block resolution
+   * is only ever needed for regular, materialized shuffles, so it always comes from the blocking
+   * manager (the only manager that produces block-manager-addressed blocks); a pipelined shuffle is
+   * served out-of-band and never resolved here. Callers that only need to resolve a block should
+   * use this rather than reaching through a `ShuffleManager`.
+   */
+  private[spark] def shuffleBlockResolver: Option[ShuffleBlockResolver] =
+    Option(_blockingShuffleManager).map(_.shuffleBlockResolver)
+
+  /**
+   * Retained for binary compatibility; returns the blocking manager. Prefer `shuffleManagerFor`
+   * (route a shuffle by its dependency) or `blockingShuffleManager` (the blocking manager
+   * explicitly), so the routing decision is explicit at the call site.
+   */
+  @deprecated("use shuffleManagerFor(dependency) to route a shuffle by type, or " +
+    "blockingShuffleManager for the blocking manager explicitly", "4.3.0")
+  def shuffleManager: ShuffleManager = _blockingShuffleManager
+
+  /**
+   * The `ShuffleManager` that serves the given shuffle, chosen by dependency type: a
+   * [[PipelinedShuffleDependency]] is served by the pipelined manager
+   * (spark.shuffle.manager.incremental, defaulting to the built-in streaming manager), every other
+   * [[ShuffleDependency]] by the blocking manager (spark.shuffle.manager). This is the single
+   * routing point for shuffle I/O; the decision is a pure function of the dependency, so the driver
+   * (at `registerShuffle`) and every executor (at `getWriter` / `getReader`) agree without any
+   * shared routing state or config re-read -- the dependency (a serialized field, deterministic on
+   * driver and executors) is always in hand at these sites, so no shuffleId -> manager tracking is
+   * needed.
+   *
+   * Whether a job may use a pipelined dependency at all is a separate, scheduler-level decision
+   * (see the fail-fast checks in `DAGScheduler`); this method only picks the implementation.
+   */
+  private[spark] def shuffleManagerFor(dependency: ShuffleDependency[_, _, _]): ShuffleManager =
+    dependency match {
+      case _: PipelinedShuffleDependency[_, _, _] =>
+        _pipelinedShuffleManager
+      case _ => _blockingShuffleManager
+    }
+
+  /**
+   * Unregisters the shuffle from every configured manager (default and, if present, incremental).
+   * Used by the `RemoveShuffle` cleanup path, which holds only a shuffleId and cannot know which
+   * manager owns it -- and must reach the owning manager on every node, including one that never
+   * performed this shuffle's I/O. Notifying all managers is safe: `unregisterShuffle` for an
+   * unknown id is a no-op. Returns true if any manager reports it removed metadata; false if no
+   * manager is initialized yet (a RemoveShuffle can arrive before the deferred init runs).
+   */
+  private[spark] def unregisterShuffleFromAllManagers(shuffleId: Int): Boolean = {
+    // Both managers are null until initializeShuffleManager runs (deferred to allow user jars), so
+    // guard them; a RemoveShuffle before init is a no-op, as it was before routing.
+    val blockingResult =
+      _blockingShuffleManager != null && _blockingShuffleManager.unregisterShuffle(shuffleId)
+    // OR the pipelined result in without short-circuiting, so both are always notified.
+    val pipelinedResult =
+      _pipelinedShuffleManager != null && _pipelinedShuffleManager.unregisterShuffle(shuffleId)
+    blockingResult || pipelinedResult
+  }
 
   /**
    * Wait for the ShuffleManager to be initialized within the specified timeout.
@@ -102,7 +193,7 @@ class SparkEnv (
    * @return true if the ShuffleManager is initialized, false otherwise
    */
   private[spark] def isShuffleManagerInitialized: Boolean = {
-    _shuffleManager != null
+    _blockingShuffleManager != null
   }
 
   // We initialize the MemoryManager later in SparkContext after DriverPlugin is loaded
@@ -185,8 +276,11 @@ class SparkEnv (
       udfDispatcherManager.foreach(_.close())
       mapOutputTracker.stop()
       _streamingShuffleOutputTracker.foreach(_.stop())
-      if (shuffleManager != null) {
-        shuffleManager.stop()
+      if (_blockingShuffleManager != null) {
+        _blockingShuffleManager.stop()
+      }
+      if (_pipelinedShuffleManager != null) {
+        _pipelinedShuffleManager.stop()
       }
       broadcastManager.stop()
       blockManager.stop()
@@ -295,19 +389,52 @@ class SparkEnv (
   }
 
   private[spark] def initializeShuffleManager(): Unit = {
-    Preconditions.checkState(null == _shuffleManager,
-      "Shuffle manager already initialized to %s", _shuffleManager)
+    Preconditions.checkState(null == _blockingShuffleManager,
+      "Shuffle manager already initialized to %s", _blockingShuffleManager)
     try {
-      _shuffleManager = ShuffleManager.create(conf, SparkContext.isDriver(executorId))
+      val isDriver = SparkContext.isDriver(executorId)
+      // The blocking manager (spark.shuffle.manager) serves all regular, materialized shuffle
+      // dependencies and owns block-by-id resolution, so it must be a BlockingShuffleManager. A
+      // manager that declares neither kind, or a pipelined-only one, is rejected here rather than
+      // silently failing block resolution later.
+      _blockingShuffleManager = ShuffleManager.create(conf, isDriver) match {
+        case blocking: BlockingShuffleManager => blocking
+        case other =>
+          throw new IllegalArgumentException(
+            s"${config.SHUFFLE_MANAGER.key} must be a BlockingShuffleManager, but " +
+              s"${other.getClass.getName} is not. A blocking manager serves regular, " +
+              s"materialized shuffles and resolves blocks by id; configure a pipelined manager " +
+              s"via ${config.SHUFFLE_MANAGER_INCREMENTAL.key} instead.")
+      }
+      // The pipelined manager (spark.shuffle.manager.incremental) serves pipelined shuffle
+      // dependencies. It is a peer of the blocking manager -- see `shuffleManagerFor` -- not a
+      // wrapper installed as the top-level manager, and is created the same way: it defaults to the
+      // built-in streaming manager (just as the blocking manager defaults to sort) and must be a
+      // PipelinedShuffleManager, since its output is read incrementally and served out-of-band and
+      // so produces no block-manager blocks.
+      _pipelinedShuffleManager = Utils.instantiateSerializerOrShuffleManager[ShuffleManager](
+        // Resolve short aliases ("sort", "tungsten-sort", "streaming") the same way the blocking
+        // manager does, so spark.shuffle.manager.incremental accepts the same values.
+        ShuffleManager.resolveShortName(conf.get(config.SHUFFLE_MANAGER_INCREMENTAL)),
+        conf, isDriver) match {
+        case pipelined: PipelinedShuffleManager => pipelined
+        case other =>
+          throw new IllegalArgumentException(
+            s"${config.SHUFFLE_MANAGER_INCREMENTAL.key} must be a PipelinedShuffleManager, but " +
+              s"${other.getClass.getName} is not. Only a pipelined manager (output read " +
+              s"incrementally and served out-of-band) belongs in this slot; a blocking manager " +
+              s"belongs in ${config.SHUFFLE_MANAGER.key}.")
+      }
     } finally {
       // Signal that the ShuffleManager has been initialized
       shuffleManagerInitLatch.countDown()
     }
     initializeStreamingShuffleOutputTracker()
   }
 
-  // Holds the streaming shuffle output tracker, which is only present when the configured
-  // shuffle manager requires it (i.e., StreamingShuffleManager or MultiShuffleManager).
+  // Holds the streaming shuffle output tracker, which is only present when the configured shuffle
+  // managers require it (i.e., a StreamingShuffleManager as the pipelined manager, or a
+  // MultiShuffleManager as the blocking manager).
   @volatile private var _streamingShuffleOutputTracker: Option[StreamingShuffleOutputTracker] =
     None
 
@@ -323,28 +450,42 @@ class SparkEnv (
       return
     }
 
-    val shuffleManagerName = ShuffleManager.getShuffleManagerClassName(conf)
-    if (shuffleManagerName == classOf[StreamingShuffleManager].getName
-        || shuffleManagerName == classOf[MultiShuffleManager].getName) {
-      val tracker = if (SparkContext.isDriver(executorId)) {
-        new StreamingShuffleOutputTrackerMaster(conf)
-      } else {
-        new StreamingShuffleOutputTrackerWorker(conf)
-      }
+    // The tracker is needed when the pipelined manager (spark.shuffle.manager.incremental) is a
+    // StreamingShuffleManager -- which is the default. Inspect the already-instantiated manager
+    // rather than re-reading the config; this runs at the end of initializeShuffleManager, so the
+    // manager is non-null here.
+    val incrementalIsStreaming =
+      _pipelinedShuffleManager.isInstanceOf[StreamingShuffleManager]
+    // It is also needed when a MultiShuffleManager is the blocking manager (spark.shuffle.manager):
+    // it internally routes some shuffles to streaming. A bare StreamingShuffleManager cannot be the
+    // blocking manager -- it is pipelined and rejected from that slot in initializeShuffleManager.
+    // TODO: remove this MultiShuffleManager clause once MultiShuffleManager is removed and
+    // streaming shuffles are served only through the incremental (pipelined) manager slot.
+    val blockingIsMulti = _blockingShuffleManager.isInstanceOf[MultiShuffleManager]
+    if (incrementalIsStreaming || blockingIsMulti) {
+      createStreamingShuffleOutputTracker()
+    }
+  }
 
-      if (SparkContext.isDriver(executorId)) {
-        tracker.trackerEndpoint = rpcEnv.setupEndpoint(
-          StreamingShuffleOutputTracker.ENDPOINT_NAME,
-          new StreamingShuffleOutputTrackerMasterEndpoint(
-            rpcEnv,
-            tracker.asInstanceOf[StreamingShuffleOutputTrackerMaster],
-            conf))
-      } else {
-        tracker.trackerEndpoint = RpcUtils.makeDriverRef(
-          StreamingShuffleOutputTracker.ENDPOINT_NAME, conf, rpcEnv)
-      }
-      _streamingShuffleOutputTracker = Some(tracker)
+  private def createStreamingShuffleOutputTracker(): Unit = {
+    val tracker = if (SparkContext.isDriver(executorId)) {
+      new StreamingShuffleOutputTrackerMaster(conf)
+    } else {
+      new StreamingShuffleOutputTrackerWorker(conf)
+    }
+
+    if (SparkContext.isDriver(executorId)) {
+      tracker.trackerEndpoint = rpcEnv.setupEndpoint(
+        StreamingShuffleOutputTracker.ENDPOINT_NAME,
+        new StreamingShuffleOutputTrackerMasterEndpoint(
+          rpcEnv,
+          tracker.asInstanceOf[StreamingShuffleOutputTrackerMaster],
+          conf))
+    } else {
+      tracker.trackerEndpoint = RpcUtils.makeDriverRef(
+        StreamingShuffleOutputTracker.ENDPOINT_NAME, conf, rpcEnv)
     }
+    _streamingShuffleOutputTracker = Some(tracker)
   }
 
   private[spark] def initializeMemoryManager(
@@ -550,7 +691,6 @@ object SparkEnv extends Logging {
             None
           }, blockManagerInfo,
           mapOutputTracker.asInstanceOf[MapOutputTrackerMaster],
-          _shuffleManager = null,
           isDriver)),
       registerOrLookupEndpoint(
         BlockManagerMaster.DRIVER_HEARTBEAT_ENDPOINT_NAME,