{"record":{"id":"0ba9183ad4047e67","repo":"apache/beam","slug":"instruction-cleaned-up-already-s","errorCode":null,"errorMessage":"Instruction cleaned up already %s","messagePattern":"Instruction cleaned up already (.+?)","errorType":"exception","errorClass":"RuntimeError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/runners/worker/data_plane.py","lineNumber":581,"sourceCode":"      self,\n      instruction_id,  # type: str\n      expected_inputs,  # type: Collection[Union[str, Tuple[str, str]]]\n      abort_callback=None  # type: Optional[Callable[[], bool]]\n  ):\n\n    # type: (...) -> Iterator[DataOrTimers]\n\n    \"\"\"\n    Generator to retrieve elements for an instruction_id\n    input_elements should be called only once for an instruction_id\n\n    Args:\n      instruction_id(str): instruction_id for which data is read\n      expected_inputs(collection): expected inputs, include both data and timer.\n    \"\"\"\n    received = self._receiving_queue(instruction_id)\n    if received is None:\n      raise RuntimeError('Instruction cleaned up already %s' % instruction_id)\n    done_inputs = set()  # type: Set[Union[str, Tuple[str, str]]]\n    abort_callback = abort_callback or (lambda: False)\n    log_interval_sec = 5 * 60\n    try:\n      start_time = time.time()\n      next_waiting_log_time = start_time + log_interval_sec\n      while len(done_inputs) < len(expected_inputs):\n        try:\n          element = received.get(timeout=1)\n        except queue.Empty:\n          if self._closed:\n            raise RuntimeError('Channel closed prematurely.')\n          if abort_callback():\n            return\n          if self._exception:\n            raise self._exception from None\n          current_time = time.time()\n          if next_waiting_log_time <= current_time:","sourceCodeStart":563,"sourceCodeEnd":599,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/runners/worker/data_plane.py#L563-L599","documentation":" 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.","triggerScenarios":" 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.","commonSituations":" 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.","solutions":["Ensure input_elements is called exactly once per instruction id, before bundle processing completes.","Check that the runner/harness does not replay or duplicate work requests with the same instruction id after cleanup.","Catch RuntimeError and treat the instruction as already processed if a duplicate read is possible.","If hit intermittently under retries, regenerate a fresh instruction id for the retried work instead of reusing it."],"exampleFix":"# before\ndata = client.input_elements(instruction_id, expected_inputs)  # may be called twice\n\n# after\nif not hasattr(seen_instructions, 'add'):\n    seen_instructions = set()\nif instruction_id in seen_instructions:\n    return  # already consumed\nseen_instructions.add(instruction_id)\ndata = client.input_elements(instruction_id, expected_inputs)","handlingStrategy":"validation","validationCode":"if instruction_id in consumed_instruction_ids:\n    raise ValueError(f'{instruction_id} already consumed; use a fresh instruction id')","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Consume each instruction id exactly once, before the bundle finalizes.","Never replay work requests with an already-cleaned instruction id.","Generate new instruction ids for retried work."],"tags":["apache-beam","data-plane","lifecycle"],"backgroundTag":"invalid-state-transition","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}