apache/beam · error · RuntimeError

Instruction cleaned up already

Error message

Instruction cleaned up already %s

What it means

Raised by GrpcClient.input_elements in the Beam FnApi data plane when the receiving queue for an instruction_id no longer exists. In-process data channels keep per-instruction queues and clean them up once consumed (instruction ids are not reusable), so reading data for an already-cleaned instruction is treated as a fatal programming/lifecycle error.

Solutions

1. Ensure input_elements is called exactly once per instruction id, before bundle processing completes.
2. Check that the runner/harness does not replay or duplicate work requests with the same instruction id after cleanup.
3. Catch RuntimeError and treat the instruction as already processed if a duplicate read is possible.
4. If hit intermittently under retries, regenerate a fresh instruction id for the retried work instead of reusing it.

Example fix

# before
data = client.input_elements(instruction_id, expected_inputs)  # may be called twice

# after
if not hasattr(seen_instructions, 'add'):
    seen_instructions = set()
if instruction_id in seen_instructions:
    return  # already consumed
seen_instructions.add(instruction_id)
data = client.input_elements(instruction_id, expected_inputs)

if instruction_id in consumed_instruction_ids:
    raise ValueError(f'{instruction_id} already consumed; use a fresh instruction id')

When it happens

Trigger: Calling input_elements(instruction_id, ...) for an instruction id whose queue was already removed by _clean_receiving_queue, typically after the bundle already finished or was aborted.

Common situations: Custom runner code or test harnesses re-reading data for a completed bundle instruction; data plane consumers racing with bundle finalization; worker restarts where instruction ids are replayed.

Understand the failure class

Background: "Invalid state transition" errors: "status must be X, actually Y", "already rejected/charging/uninstalled", "cannot ... while running" — what they mean when a library rejects your call — this error's family across 31 libraries.

Related errors

• At least one owner must be registered.
• Buffer counter not initialized for UUID:
• Bundle is not in a finalizable state for
• Cannot access FinishBundleContext outside of @FinishBundle…
• Cannot access StartBundleContext outside of @StartBundle…

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/0ba9183ad4047e67. Report an issue: GitHub.

Appendix: source

Thrown at sdks/python/apache_beam/runners/worker/data_plane.py:581

      self,
      instruction_id,  # type: str
      expected_inputs,  # type: Collection[Union[str, Tuple[str, str]]]
      abort_callback=None  # type: Optional[Callable[[], bool]]
  ):

    # type: (...) -> Iterator[DataOrTimers]

    """
    Generator to retrieve elements for an instruction_id
    input_elements should be called only once for an instruction_id

    Args:
      instruction_id(str): instruction_id for which data is read
      expected_inputs(collection): expected inputs, include both data and timer.
    """
    received = self._receiving_queue(instruction_id)
    if received is None:
      raise RuntimeError('Instruction cleaned up already %s' % instruction_id)
    done_inputs = set()  # type: Set[Union[str, Tuple[str, str]]]
    abort_callback = abort_callback or (lambda: False)
    log_interval_sec = 5 * 60
    try:
      start_time = time.time()
      next_waiting_log_time = start_time + log_interval_sec
      while len(done_inputs) < len(expected_inputs):
        try:
          element = received.get(timeout=1)
        except queue.Empty:
          if self._closed:
            raise RuntimeError('Channel closed prematurely.')
          if abort_callback():
            return
          if self._exception:
            raise self._exception from None
          current_time = time.time()
          if next_waiting_log_time <= current_time:

View on GitHub (pinned to 12126d8942)