{"record":{"id":"bd12395c1d02937c","repo":"apache/beam","slug":"unable-to-handle-state-requests-for-processbundledescriptor","errorCode":null,"errorMessage":"Unable to handle state requests for ProcessBundleDescriptor for bundle id %s.","messagePattern":"Unable to handle state requests for ProcessBundleDescriptor for bundle id (.+?)\\.","errorType":"exception","errorClass":"RuntimeError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/runners/worker/sdk_worker.py","lineNumber":1003,"sourceCode":"    raise NotImplementedError(type(self))\n\n  @abc.abstractmethod\n  def clear(self, state_key):\n    # type: (beam_fn_api_pb2.StateKey) -> _Future\n    raise NotImplementedError(type(self))\n\n  @abc.abstractmethod\n  def done(self):\n    # type: () -> None\n    raise NotImplementedError(type(self))\n\n\nclass ThrowingStateHandler(CachingStateHandler):\n  \"\"\"A caching state handler that errors on any requests.\"\"\"\n  @contextlib.contextmanager\n  def process_instruction_id(self, bundle_id, cache_tokens):\n    # type: (str, Iterable[beam_fn_api_pb2.ProcessBundleRequest.CacheToken]) -> Iterator[None]\n    raise RuntimeError(\n        'Unable to handle state requests for ProcessBundleDescriptor '\n        'for bundle id %s.' % bundle_id)\n\n  def blocking_get(\n      self,\n      state_key,  # type: beam_fn_api_pb2.StateKey\n      coder,  # type: coder_impl.CoderImpl\n  ):\n    # type: (...) -> Iterable[Any]\n    raise RuntimeError(\n        'Unable to handle state requests for ProcessBundleDescriptor without '\n        'state ApiServiceDescriptor for state key %s.' % state_key)\n\n  def extend(\n      self,\n      state_key,  # type: beam_fn_api_pb2.StateKey\n      coder,  # type: coder_impl.CoderImpl\n      elements,  # type: Iterable[Any]","sourceCodeStart":985,"sourceCodeEnd":1021,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/runners/worker/sdk_worker.py#L985-L1021","documentation":"This RuntimeError is thrown by ThrowingStateHandler, a CachingStateHandler subclass in the Beam SDK worker that deliberately errors on every state request. It is used in tests or fallback paths where a bundle requests state operations but the harness has no functioning state API service configured. The contextmanager form fails immediately when a bundle's process_instruction_id is entered.","triggerScenarios":"A worker is constructed with ThrowingStateHandler (no state ApiServiceDescriptor wired) and the runner sends a ProcessBundleRequest whose instructions enter process_instruction_id(bundle_id, cache_tokens).","commonSituations":"Test harnesses or custom workers where stateful (user-state / bag-state) transforms are executed without a state gRPC service; pipeline configurations that enable state but launch a stateless harness.","solutions":["Provide a real state handler (GrpcStateHandler/CachingStateHandler bound to the state ApiServiceDescriptor) when constructing the SDK harness","Remove or rewrite stateful transforms (DoFns using beam.state) from pipelines run against this worker","If this occurs in tests, ensure the test provisions a state ApiServiceDescriptor before process_bundle"],"exampleFix":"// before\nstate_handler = ThrowingStateHandler()\n// after\nstate_handler = CachingStateHandler(state_api_service_descriptor)","handlingStrategy":"try-catch","validationCode":"assert isinstance(state_handler, (GrpcStateHandler, CachingStateHandler)), 'state handler must serve state requests'","typeGuard":"def has_state_service(handler): return not isinstance(handler, ThrowingStateHandler)","tryCatchPattern":"try:\n  with handler.process_instruction_id(bundle_id, cache_tokens):\n    ...\nexcept RuntimeError as e:\n  log.error('state handling unavailable: %s', e)\n  raise","preventionTips":["Always wire a GrpcStateHandler when pipelines use beam.state","Include state ApiServiceDescriptor in ProcessBundleDescriptors","Add a startup check that the state handler is not a stub"],"tags":["apache-beam","state-handler","worker","runtime"],"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-20T03:17:13.778Z"}