apache/beam · error · RuntimeError

All workers communicate through gRPC should have worker_id.

Error message

All workers communicate through gRPC should have worker_id. Received None.

What it means

The gRPC control handler for a worker requires every incoming RPC stream to carry a 'worker_id' entry in its invocation metadata so the server can route the stream to the right control connection. When the metadata key is absent, the handler raises this RuntimeError instead of proceeding. It exists because Beam's Fn API multiplexes multiple workers over one gRPC server and cannot correlate streams without the id.

Source

Thrown at sdks/python/apache_beam/runners/portability/fn_api_runner/worker_handlers.py:222

    # type: (str) -> ControlConnection
    with self._lock:
      return self._connections_by_worker_id[worker_id]

  def Control(
      self,
      iterator,  # type: Iterable[beam_fn_api_pb2.InstructionResponse]
      context  # type: ServicerContext
  ):
    # type: (...) -> Iterator[beam_fn_api_pb2.InstructionRequest]
    with self._lock:
      if self._state == self.DONE_STATE:
        return
      else:
        self._state = self.STARTED_STATE

    worker_id = dict(context.invocation_metadata()).get('worker_id')
    if not worker_id:
      raise RuntimeError(
          'All workers communicate through gRPC should have '
          'worker_id. Received None.')

    control_conn = self.get_conn_by_worker_id(worker_id)
    control_conn.set_input(iterator)

    while True:
      to_push = control_conn.get_req()
      if to_push is self._DONE_MARKER:
        return
      yield to_push
      if self._log_req:
        self._req_sent[to_push.instruction_id] += 1

  def done(self):
    # type: () -> None
    self._state = self.DONE_STATE
    _LOGGER.debug(

View on GitHub (pinned to 12126d8942)

Solutions

1. Use a worker/SDK container image whose version matches the Beam runner version (check --sdk_harness_container_image_overrides).
2. If writing a custom client, add ('worker_id', '<id>') to the gRPC invocation metadata of the control RPC.
3. Remove custom gRPC interceptors/proxies that drop unknown metadata keys.
4. Ensure the worker connects via the provision/log endpoints normally (start_worker) rather than being pointed at the control port directly.

Example fix

// before (custom harness client)
stub.Control(iter_requests)
// after
metadata = [('worker_id', worker_id)]
stub.Control(iter_requests, metadata=metadata)

md = dict(invocation_metadata)
if not md.get('worker_id'):
    raise ValueError('control RPC must include worker_id metadata')

def has_worker_id(md): return bool(dict(md).get('worker_id'))

When it happens

Trigger: A worker SDK harness (or custom/manual gRPC client) opens the BeamFnControl stream without setting 'worker_id' in the invocation metadata, e.g. when launching a container or harness manually with an incompatible/older SDK image or custom worker entry point.

Common situations: Mixing Beam SDK versions (older harness image that doesn't send worker_id) with a newer runner; custom container images overriding the harness entry point; hand-written gRPC clients talking to the Fn API control port; proxying/gRPC middleware stripping metadata.

Understand the failure class

Background: "missing required argument" and "the following required arguments were not provided": what required-argument errors mean and how to fix them — this error's family across 20 libraries.

Related errors

• State stream is closed.
• Unknown type
• Instruction id was registered twice
• Logging stream terminated unexpectedly with success before i
• Failed to get table schema for table:

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