apache/beam · critical · RuntimeError

Channel closed prematurely.

Error message

Channel closed prematurely.

What it means

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.

Source

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

    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:
            # If at the same time another instruction is waiting on input queue
            # to become available, it is a sign of inefficiency in data plane.
            _LOGGER.info(
                'Detected input queue delay longer than %s seconds. '
                'Waiting to receive elements in input queue '
                'for instruction: %s for %.2f seconds.',
                log_interval_sec,
                instruction_id,
                current_time - start_time)
            next_waiting_log_time = current_time + log_interval_sec
        else:
          start_time = time.time()

View on GitHub (pinned to 12126d8942)

Solutions

1. Check SDK harness worker logs near the failure for crashes (OOM, SIGKILL) and fix the root resource issue.
2. Increase gRPC/stream timeouts and keep-alive settings between runner and worker if the link is being dropped.
3. Improve network reliability (same-AZ worker placement, retry policies) in cluster configuration.
4. Retry the failed bundle; the runner should re-execute the work since the channel cannot be recovered.

try:
    for el in client.input_elements(instruction_id, expected):
        handle(el)
except RuntimeError as e:
    if 'Channel closed prematurely' in str(e):
        schedule_bundle_retry(instruction_id)  # channel unrecoverable
    raise

When it happens

Trigger: 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.

Common situations: 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.

Related errors

• error connecting to job server at %v: %v
• Error starting background log thread
• No RabbitMQ channel available
• Failed to connect to host: %s, error: %s
• input is already set.

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