class_docstr: Shuffle plugin for the scheduler
  This coordinates the individual worker plugins to ensure correctness
  and collects heartbeat messages for the dashboard.
  See Also
  --------
  ShuffleWorkerPlugin
class_docstr_tok: ["Shuffle","plugin","for","the","scheduler","This","coordinates","the","individual","worker","plugins","to","ensure","correctness","and","collects","heartbeat","messages","for","the","dashboard",".","See","Also","--","--","--","--","ShuffleWorkerPlugin"]
code: def _restart_recommendations(self, id: ShuffleId) -> Recs:
      barrier_task = self.scheduler.tasks[barrier_key(id)]
      recs: Recs = {}
  
      for dt in barrier_task.dependents:
          if dt.state == "erred":
              return {}
          recs.update({dt.key: "released"})
  
      if barrier_task.state == "erred":
          # This should never happen, a dependent of the barrier should already
          # be `erred`
          raise RuntimeError(
              f"Expected dependents of {barrier_task=} to be 'erred' if "
              "the barrier is."
          )  # pragma: no cover
      recs.update({barrier_task.key: "released"})
  
      for dt in barrier_task.dependencies:
          if dt.state == "erred":
              # This should never happen, a dependent of the barrier should already
              # be `erred`
              raise RuntimeError(
                  f"Expected barrier and its dependents to be "
                  f"'erred' if the barrier's dependency {dt} is."
              )  # pragma: no cover
          recs.update({dt.key: "released"})
      return recs
  
imports: from __future__ import annotations
  import contextlib
  import itertools
  import logging
  from collections import defaultdict
  from typing import TYPE_CHECKING, Any
  from dask.typing import Key
  from distributed.diagnostics.plugin import SchedulerPlugin
  from distributed.metrics import time
  from distributed.protocol.pickle import dumps
  from distributed.protocol.serialize import ToPickle
  from distributed.shuffle._core import SchedulerShuffleState, ShuffleId, ShuffleRunSpec, ShuffleSpec, barrier_key, id_from_key
  from distributed.shuffle._worker_plugin import ShuffleWorkerPlugin
  from distributed.utils import log_errors
label_desc: Use image node_id 13 for calling the ShuffleSchedulerPlugin obj's underlying member method code with example usage: obj._restart_recommendations(id) and returns: recs, dict