{"record":{"id":"0f270df12715ed6b","repo":"apache/beam","slug":"already-bound-to-r","errorCode":null,"errorMessage":"Already bound to %r","messagePattern":"Already bound to %r","errorType":"exception","errorClass":"RuntimeError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/runners/worker/sdk_worker.py","lineNumber":1060,"sourceCode":"  _DONE = Sentinel.sentinel\n\n  def __init__(self, state_stub):\n    # type: (beam_fn_api_pb2_grpc.BeamFnStateStub) -> None\n    self._lock = threading.Lock()\n    self._state_stub = state_stub\n    self._requests = queue.Queue(\n    )  # type: queue.Queue[Union[beam_fn_api_pb2.StateRequest, Sentinel]]\n    self._responses_by_id = {}  # type: Dict[str, _Future]\n    self._last_id = 0\n    self._exception = None  # type: Optional[Exception]\n    self._context = threading.local()\n    self.start()\n\n  @contextlib.contextmanager\n  def process_instruction_id(self, bundle_id):\n    # type: (str) -> Iterator[None]\n    if getattr(self._context, 'process_instruction_id', None) is not None:\n      raise RuntimeError(\n          'Already bound to %r' % self._context.process_instruction_id)\n    self._context.process_instruction_id = bundle_id\n    try:\n      yield\n    finally:\n      self._context.process_instruction_id = None\n\n  def start(self):\n    # type: () -> None\n    self._done = False\n\n    def request_iter():\n      # type: () -> Iterator[beam_fn_api_pb2.StateRequest]\n      while True:\n        request = self._requests.get()\n        if request is self._DONE or self._done:\n          break\n        yield request","sourceCodeStart":1042,"sourceCodeEnd":1078,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/runners/worker/sdk_worker.py#L1042-L1078","documentation":"GrpcStateHandler.process_instruction_id is a contextmanager that binds exactly one bundle id to the thread's request context at a time. If a second bundle is entered while one is still bound (never exited), it raises 'Already bound to' the stale id, protecting against overlapping bundle execution on one context.","triggerScenarios":"Nesting or missing exit of the process_instruction_id contextmanager: an exception path skips the finally-unbind, or two bundles are processed concurrently on the same handler/context.","commonSituations":"Buggy custom runners reusing a handler across bundles; callback/exception leaks where the context manager wasn't exited; shared thread-local context misuse.","solutions":["Ensure every `with handler.process_instruction_id(id):` block is properly exited (avoid swallowing BaseException between enter and exit)","Create a fresh GrpcStateHandler per worker thread rather than sharing one","Check for nested process_instruction_id calls and flatten them"],"exampleFix":"// before\nhandler.process_instruction_id(bundle_a).__enter__()\nhandler.process_instruction_id(bundle_b).__enter__()  # raises\n// after\nwith handler.process_instruction_id(bundle_a):\n  ...\nwith handler.process_instruction_id(bundle_b):\n  ...","handlingStrategy":"try-catch","validationCode":"if getattr(handler._context, 'process_instruction_id', None) is not None: raise StateError('bundle already active on this handler')","typeGuard":"def bundle_active(handler): return getattr(handler._context, 'process_instruction_id', None) is not None","tryCatchPattern":"try:\n  with handler.process_instruction_id(bundle_id):\n    ...\nexcept RuntimeError as e:\n  if str(e).startswith('Already bound'):\n    log.error('overlapping bundle execution: %s', e)\n  raise","preventionTips":["Never nest or reuse process_instruction_id contexts","One bundle per handler/thread; reset thread-locals on exception","Use contextlib.ExitStack for guaranteed unbind"],"tags":["apache-beam","grpc","state-handler","concurrency"],"backgroundTag":"invalid-state-transition","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"}