apache/beam · error · RuntimeError

Already bound to %r

Error message

Already bound to %r

What it means

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.

Solutions

1. Ensure every `with handler.process_instruction_id(id):` block is properly exited (avoid swallowing BaseException between enter and exit)
2. Create a fresh GrpcStateHandler per worker thread rather than sharing one
3. Check for nested process_instruction_id calls and flatten them

Example fix

// before
handler.process_instruction_id(bundle_a).__enter__()
handler.process_instruction_id(bundle_b).__enter__()  # raises
// after
with handler.process_instruction_id(bundle_a):
  ...
with handler.process_instruction_id(bundle_b):
  ...

if getattr(handler._context, 'process_instruction_id', None) is not None: raise StateError('bundle already active on this handler')

def bundle_active(handler): return getattr(handler._context, 'process_instruction_id', None) is not None

try:
  with handler.process_instruction_id(bundle_id):
    ...
except RuntimeError as e:
  if str(e).startswith('Already bound'):
    log.error('overlapping bundle execution: %s', e)
  raise

When it happens

Trigger: 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.

Common situations: Buggy custom runners reusing a handler across bundles; callback/exception leaks where the context manager wasn't exited; shared thread-local context misuse.

Understand the failure class

Background: "Invalid state transition" errors: "status must be X, actually Y", "already rejected/charging/uninstalled", "cannot ... while running" — what they mean when a library rejects your call — this error's family across 31 libraries.

Related errors

• Cache tokens already set to
• Buffer size exceeds GRPC limit . This is likely due to a…
• Cannot interpret a request received over control channel…
• Cannot schedule new tasks after thread pool has been…
• Channel closed prematurely.

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

Appendix: source

Thrown at sdks/python/apache_beam/runners/worker/sdk_worker.py:1060

  _DONE = Sentinel.sentinel

  def __init__(self, state_stub):
    # type: (beam_fn_api_pb2_grpc.BeamFnStateStub) -> None
    self._lock = threading.Lock()
    self._state_stub = state_stub
    self._requests = queue.Queue(
    )  # type: queue.Queue[Union[beam_fn_api_pb2.StateRequest, Sentinel]]
    self._responses_by_id = {}  # type: Dict[str, _Future]
    self._last_id = 0
    self._exception = None  # type: Optional[Exception]
    self._context = threading.local()
    self.start()

  @contextlib.contextmanager
  def process_instruction_id(self, bundle_id):
    # type: (str) -> Iterator[None]
    if getattr(self._context, 'process_instruction_id', None) is not None:
      raise RuntimeError(
          'Already bound to %r' % self._context.process_instruction_id)
    self._context.process_instruction_id = bundle_id
    try:
      yield
    finally:
      self._context.process_instruction_id = None

  def start(self):
    # type: () -> None
    self._done = False

    def request_iter():
      # type: () -> Iterator[beam_fn_api_pb2.StateRequest]
      while True:
        request = self._requests.get()
        if request is self._DONE or self._done:
          break
        yield request

View on GitHub (pinned to 12126d8942)