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"]
docstr: Restart all active shuffles when a participating worker leaves the cluster.
  
  .. note::
      Due to the order of operations in :meth:`~Scheduler.remove_worker`, the
      shuffle may have already been archived by
      :meth:`~ShuffleSchedulerPlugin.transition`. In this case, the
      ``stimulus_id`` is used as a transaction identifier and all archived shuffles
      with a matching `stimulus_id` are restarted.
docstr_tok: ["Restart","all","active","shuffles","when","a","participating","worker","leaves","the","cluster",".","..","note",":",":","Due","to","the","order","of","operations","in",":","meth",":","`","~Scheduler.remove_worker","`",",","the","shuffle","may","have","already","been","archived","by",":","meth",":","`","~ShuffleSchedulerPlugin.transition","`",".","In","this","case",",","the","``","stimulus_id","``","is","used","as","a","transaction","identifier","and","all","archived","shuffles","with","a","matching","`","stimulus_id","`","are","restarted","."]
code: def remove_worker(
      self,
      scheduler: Scheduler,
      worker: str,
      *,
      stimulus_id: str,
      **kwargs: Any,
  ) -> None:
      """Restart all active shuffles when a participating worker leaves the cluster.
  
      .. note::
          Due to the order of operations in :meth:`~Scheduler.remove_worker`, the
          shuffle may have already been archived by
          :meth:`~ShuffleSchedulerPlugin.transition`. In this case, the
          ``stimulus_id`` is used as a transaction identifier and all archived shuffles
          with a matching `stimulus_id` are restarted.
      """
  
      # If processing the transactions causes a task to get released, this
      # removes the shuffle from self.active_shuffles. Therefore, we must iterate
      # over a copy.
      for shuffle_id, shuffle in self.active_shuffles.copy().items():
          if worker not in shuffle.participating_workers:
              continue
          logger.debug(
              "Worker %s removed during active shuffle %s due to stimulus '%s'",
              worker,
              shuffle_id,
              stimulus_id,
          )
          exception = RuntimeError(
              f"Worker {worker} left during active {shuffle}"
          )
          self._fail_on_workers(shuffle, str(exception))
          self._clean_on_scheduler(shuffle_id, stimulus_id)
  
      for shuffle in self._archived_by_stimulus.get(stimulus_id, set()):
          self._restart_shuffle(
              shuffle.id, scheduler, stimulus_id=stimulus_id
          )
  
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 15 for calling the ShuffleSchedulerPlugin obj's underlying member method code with example usage: obj.remove_worker(scheduler, worker) without return types