class_docstr: Interface between a Worker and a Shuffle.
  
  This extension is responsible for
  
  - Lifecycle of Shuffle instances
  - ensuring connectivity between remote shuffle instances
  - ensuring connectivity and integration with the scheduler
  - routing concurrent calls to the appropriate `Shuffle` based on its `ShuffleID`
  - collecting instrumentation of ongoing shuffles and route to scheduler/worker
class_docstr_tok: ["Interface","between","a","Worker","and","a","Shuffle",".","This","extension","is","responsible","for","-","Lifecycle","of","Shuffle","instances","-","ensuring","connectivity","between","remote","shuffle","instances","-","ensuring","connectivity","and","integration","with","the","scheduler","-","routing","concurrent","calls","to","the","appropriate","`","Shuffle","`","based","on","its","`","ShuffleID","`","-","collecting","instrumentation","of","ongoing","shuffles","and","route","to","scheduler\/worker"]
docstr: Task: Retrieve a shuffled output partition from the ShuffleWorkerPlugin.
  
  Calling this for a ``shuffle_id`` which is unknown or incomplete is an error.
docstr_tok: ["Task",":","Retrieve","a","shuffled","output","partition","from","the","ShuffleWorkerPlugin",".","Calling","this","for","a","``","shuffle_id","``","which","is","unknown","or","incomplete","is","an","error","."]
code: def get_output_partition(
      self,
      shuffle_id: ShuffleId,
      run_id: int,
      partition_id: int | NDIndex,
      meta: pd.DataFrame | None = None,
  ) -> Any:
      """
      Task: Retrieve a shuffled output partition from the ShuffleWorkerPlugin.
  
      Calling this for a ``shuffle_id`` which is unknown or incomplete is an error.
      """
      shuffle_run = self.get_shuffle_run(shuffle_id, run_id)
      key = thread_state.key
      return shuffle_run.get_output_partition(
          partition_id=partition_id,
          key=key,
          meta=meta,
      )
  
imports: from __future__ import annotations
  import asyncio
  import logging
  from collections.abc import Sequence
  from concurrent.futures import ThreadPoolExecutor
  from typing import TYPE_CHECKING, Any, overload
  from dask.context import thread_state
  from dask.typing import Key
  from dask.utils import parse_bytes
  from distributed.diagnostics.plugin import WorkerPlugin
  from distributed.protocol.serialize import ToPickle
  from distributed.shuffle._core import NDIndex, ShuffleId, ShuffleRun, ShuffleRunSpec, ShuffleSpec
  from distributed.shuffle._exceptions import ShuffleClosedError
  from distributed.shuffle._limiter import ResourceLimiter
  from distributed.utils import log_errors, sync
label_desc: Use image node_id 10 for calling the ShuffleWorkerPlugin obj's underlying member method code with example usage: obj.get_output_partition(shuffle_id, run_id, partition_id, meta) and returns: shuffle_run