apache/beam · error · RuntimeError

Header metadata already has a worker_id.

Error message

Header metadata already has a worker_id.

What it means

WorkerIdInterceptor is a gRPC client interceptor that injects the worker's id into every outgoing request's metadata. It raises RuntimeError if the metadata already contains a worker_id entry, because a duplicate or conflicting worker_id would break header uniqueness guarantees.

Source

Thrown at sdks/python/apache_beam/runners/worker/worker_id_interceptor.py:65

  def intercept_unary_unary(self, continuation, client_call_details, request):
    return self._intercept(continuation, client_call_details, request)

  def intercept_unary_stream(self, continuation, client_call_details, request):
    return self._intercept(continuation, client_call_details, request)

  def intercept_stream_unary(self, continuation, client_call_details, request):
    return self._intercept(continuation, client_call_details, request)

  def intercept_stream_stream(
      self, continuation, client_call_details, request_iterator):
    return self._intercept(continuation, client_call_details, request_iterator)

  def _intercept(self, continuation, client_call_details, request):
    metadata = []
    if client_call_details.metadata is not None:
      metadata = list(client_call_details.metadata)
    if 'worker_id' in metadata:
      raise RuntimeError('Header metadata already has a worker_id.')
    metadata.append(('worker_id', self._worker_id))
    new_client_details = _ClientCallDetails(
        client_call_details.method,
        client_call_details.timeout,
        metadata,
        client_call_details.credentials)
    return continuation(new_client_details, request)

View on GitHub (pinned to 12126d8942)

Solutions

1. Remove any explicit worker_id from the request metadata and let the interceptor add it
2. Ensure the interceptor is attached to a channel only once; do not wrap already-intercepted stubs
3. If you need a custom worker_id, set it via the interceptor's constructor rather than per-call metadata

Example fix

// before
stub = create_stub(channel, extra_metadata=[('worker_id', my_id)])  # duplicate worker_id
// after
stub = create_stub(channel)  # interceptor injects worker_id once

def metadata_has_no_worker_id(metadata):
    return metadata is None or all(k != 'worker_id' for k, _ in metadata)

try:
    response = stub.Call(request, metadata=metadata)
except RuntimeError as e:
    if 'Header metadata already has a worker_id' in str(e):
        response = stub.Call(request, metadata=[m for m in (metadata or []) if m[0] != 'worker_id'])
    else:
        raise

When it happens

Trigger: Applying the interceptor twice to the same gRPC channel, or a caller explicitly passing metadata containing a worker_id entry (e.g. (('worker_id', 'x'),)) on a stub wrapped by this interceptor via intercept_unary_unary / intercept_unary_stream / intercept_stream_unary / intercept_stream_stream.

Common situations: Building channels with nested interceptors in Beam worker startup code; custom client code adding worker_id manually while Beam's interceptor also adds it; misconfigured channel factories that re-wrap already-intercepted channels.

Understand the failure class

Background: "This is a bug, please report it": internal invariant violations, unreachable panics, and SNH errors explained — this error's family across 47 libraries.

Related errors

• input is already set.
• Buffer size {size} exceeds GRPC limit {_FLUSH_MAX_SIZE}. Thi
• Service failed to start up with error %s
• Transform service did not start in %s seconds.
• failed to read metadata from context

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