{"record":{"id":"c79f74ec09fde1a2","repo":"apache/beam","slug":"all-workers-communicate-through-grpc-should-have-worker-id","errorCode":null,"errorMessage":"All workers communicate through gRPC should have worker_id. Received None.","messagePattern":"All workers communicate through gRPC should have worker_id\\. Received None\\.","errorType":"exception","errorClass":"RuntimeError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/runners/portability/fn_api_runner/worker_handlers.py","lineNumber":222,"sourceCode":"    # type: (str) -> ControlConnection\n    with self._lock:\n      return self._connections_by_worker_id[worker_id]\n\n  def Control(\n      self,\n      iterator,  # type: Iterable[beam_fn_api_pb2.InstructionResponse]\n      context  # type: ServicerContext\n  ):\n    # type: (...) -> Iterator[beam_fn_api_pb2.InstructionRequest]\n    with self._lock:\n      if self._state == self.DONE_STATE:\n        return\n      else:\n        self._state = self.STARTED_STATE\n\n    worker_id = dict(context.invocation_metadata()).get('worker_id')\n    if not worker_id:\n      raise RuntimeError(\n          'All workers communicate through gRPC should have '\n          'worker_id. Received None.')\n\n    control_conn = self.get_conn_by_worker_id(worker_id)\n    control_conn.set_input(iterator)\n\n    while True:\n      to_push = control_conn.get_req()\n      if to_push is self._DONE_MARKER:\n        return\n      yield to_push\n      if self._log_req:\n        self._req_sent[to_push.instruction_id] += 1\n\n  def done(self):\n    # type: () -> None\n    self._state = self.DONE_STATE\n    _LOGGER.debug(","sourceCodeStart":204,"sourceCodeEnd":240,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/runners/portability/fn_api_runner/worker_handlers.py#L204-L240","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Use a worker/SDK container image whose version matches the Beam runner version (check --sdk_harness_container_image_overrides).","If writing a custom client, add ('worker_id', '<id>') to the gRPC invocation metadata of the control RPC.","Remove custom gRPC interceptors/proxies that drop unknown metadata keys.","Ensure the worker connects via the provision/log endpoints normally (start_worker) rather than being pointed at the control port directly."],"exampleFix":"// before (custom harness client)\nstub.Control(iter_requests)\n// after\nmetadata = [('worker_id', worker_id)]\nstub.Control(iter_requests, metadata=metadata)","handlingStrategy":"validation","validationCode":"md = dict(invocation_metadata)\nif not md.get('worker_id'):\n    raise ValueError('control RPC must include worker_id metadata')","typeGuard":"def has_worker_id(md): return bool(dict(md).get('worker_id'))","tryCatchPattern":null,"preventionTips":["Keep SDK harness images version-matched to the runner","Never strip gRPC metadata in interceptors/proxies","Always launch workers through the runner's start_worker path"],"tags":["grpc","worker-lifecycle","metadata","beam"],"backgroundTag":"missing-required-argument","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"}