{"record":{"id":"f16119f94a0a050e","repo":"apache/beam","slug":"channel-closed-prematurely","errorCode":null,"errorMessage":"Channel closed prematurely.","messagePattern":"Channel closed prematurely\\.","errorType":"exception","errorClass":"RuntimeError","httpStatus":null,"severity":"critical","filePath":"sdks/python/apache_beam/runners/worker/data_plane.py","lineNumber":593,"sourceCode":"    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:\n            # If at the same time another instruction is waiting on input queue\n            # to become available, it is a sign of inefficiency in data plane.\n            _LOGGER.info(\n                'Detected input queue delay longer than %s seconds. '\n                'Waiting to receive elements in input queue '\n                'for instruction: %s for %.2f seconds.',\n                log_interval_sec,\n                instruction_id,\n                current_time - start_time)\n            next_waiting_log_time = current_time + log_interval_sec\n        else:\n          start_time = time.time()","sourceCodeStart":575,"sourceCodeEnd":611,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/runners/worker/data_plane.py#L575-L611","documentation":" Raised inside input_elements while waiting on the per-instruction queue: the wait timed out (queue.Empty) and the data channel was marked closed (self._closed). It means the gRPC data stream ended before all expected inputs (data/timers) were delivered for the instruction.","triggerScenarios":" input_elements loops on received.get(timeout=1) waiting for elements for each expected input; the runner-side channel closes (worker shutdown, stream reset, network drop) before an is_last marker arrives.","commonSituations":" Worker process killed mid-bundle (OOM, preemption); network instability between runner and SDK harness; runner shutting down while a bundle still reads inputs; gRPC stream failures from timeouts.","solutions":["Check SDK harness worker logs near the failure for crashes (OOM, SIGKILL) and fix the root resource issue.","Increase gRPC/stream timeouts and keep-alive settings between runner and worker if the link is being dropped.","Improve network reliability (same-AZ worker placement, retry policies) in cluster configuration.","Retry the failed bundle; the runner should re-execute the work since the channel cannot be recovered."],"exampleFix":null,"handlingStrategy":"retry","validationCode":null,"typeGuard":null,"tryCatchPattern":"try:\n    for el in client.input_elements(instruction_id, expected):\n        handle(el)\nexcept RuntimeError as e:\n    if 'Channel closed prematurely' in str(e):\n        schedule_bundle_retry(instruction_id)  # channel unrecoverable\n    raise","preventionTips":["Monitor worker memory; OOM kills are the top cause of premature channel closure.","Configure gRPC keepalive and generous deadlines on runner-worker channels.","Keep workers network-proximate to the runner; rely on bundle-level retries."],"tags":["apache-beam","grpc","data-plane","connection"],"backgroundTag":"broken-pipe","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}