{"record":{"id":"98ef58ea57567ba7","repo":"apache/beam","slug":"header-metadata-already-has-a-worker-id","errorCode":null,"errorMessage":"Header metadata already has a worker_id.","messagePattern":"Header metadata already has a worker_id\\.","errorType":"exception","errorClass":"RuntimeError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/runners/worker/worker_id_interceptor.py","lineNumber":65,"sourceCode":"  def intercept_unary_unary(self, continuation, client_call_details, request):\n    return self._intercept(continuation, client_call_details, request)\n\n  def intercept_unary_stream(self, continuation, client_call_details, request):\n    return self._intercept(continuation, client_call_details, request)\n\n  def intercept_stream_unary(self, continuation, client_call_details, request):\n    return self._intercept(continuation, client_call_details, request)\n\n  def intercept_stream_stream(\n      self, continuation, client_call_details, request_iterator):\n    return self._intercept(continuation, client_call_details, request_iterator)\n\n  def _intercept(self, continuation, client_call_details, request):\n    metadata = []\n    if client_call_details.metadata is not None:\n      metadata = list(client_call_details.metadata)\n    if 'worker_id' in metadata:\n      raise RuntimeError('Header metadata already has a worker_id.')\n    metadata.append(('worker_id', self._worker_id))\n    new_client_details = _ClientCallDetails(\n        client_call_details.method,\n        client_call_details.timeout,\n        metadata,\n        client_call_details.credentials)\n    return continuation(new_client_details, request)\n","sourceCodeStart":47,"sourceCodeEnd":73,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/runners/worker/worker_id_interceptor.py#L47-L73","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Remove any explicit worker_id from the request metadata and let the interceptor add it","Ensure the interceptor is attached to a channel only once; do not wrap already-intercepted stubs","If you need a custom worker_id, set it via the interceptor's constructor rather than per-call metadata"],"exampleFix":"// before\nstub = create_stub(channel, extra_metadata=[('worker_id', my_id)])  # duplicate worker_id\n// after\nstub = create_stub(channel)  # interceptor injects worker_id once","handlingStrategy":"try-catch","validationCode":"def metadata_has_no_worker_id(metadata):\n    return metadata is None or all(k != 'worker_id' for k, _ in metadata)","typeGuard":null,"tryCatchPattern":"try:\n    response = stub.Call(request, metadata=metadata)\nexcept RuntimeError as e:\n    if 'Header metadata already has a worker_id' in str(e):\n        response = stub.Call(request, metadata=[m for m in (metadata or []) if m[0] != 'worker_id'])\n    else:\n        raise","preventionTips":["Do not set worker_id manually in request metadata on intercepted channels","Attach WorkerIdInterceptor to each channel exactly once","Keep channel/stub construction centralized to avoid double-wrapping"],"tags":["python","apache-beam","grpc","interceptor","metadata"],"backgroundTag":"internal-invariant-violation","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"}