{"record":{"id":"62d1124b8ec77068","repo":"apache/beam","slug":"cache-tokens-already-set-to-s","errorCode":null,"errorMessage":"Cache tokens already set to %s","messagePattern":"Cache tokens already set to (.+?)","errorType":"exception","errorClass":"RuntimeError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/runners/worker/sdk_worker.py","lineNumber":1196,"sourceCode":"      self,\n      global_state_cache,  # type: StateCache\n      underlying_state  # type: StateHandler\n  ):\n    # type: (...) -> None\n    self._underlying = underlying_state\n    self._state_cache = global_state_cache\n    self._context = threading.local()\n\n    # state retrieval time statistics\n    self._retrieval_time = 0.0\n    self._get_raw_called = 0\n    self._warn_interval = 60.0\n\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    if getattr(self._context, 'user_state_cache_token', None) is not None:\n      raise RuntimeError(\n          'Cache tokens already set to %s' %\n          self._context.user_state_cache_token)\n    self._context.side_input_cache_tokens = {}\n    user_state_cache_token = None\n    for cache_token_struct in cache_tokens:\n      if cache_token_struct.HasField(\"user_state\"):\n        # There should only be one user state token present\n        assert not user_state_cache_token\n        user_state_cache_token = cache_token_struct.token\n      elif cache_token_struct.HasField(\"side_input\"):\n        self._context.side_input_cache_tokens[\n            cache_token_struct.side_input.transform_id,\n            cache_token_struct.side_input.\n            side_input_id] = cache_token_struct.token\n    # TODO: Consider a two-level cache to avoid extra logic and locking\n    # for items cached at the bundle level.\n    self._context.bundle_cache_token = bundle_id\n    try:","sourceCodeStart":1178,"sourceCodeEnd":1214,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/runners/worker/sdk_worker.py#L1178-L1214","documentation":"CachingStateHandler.process_instruction_id sets the per-instruction cache tokens (user_state_cache_token and side_input_cache_tokens). If the thread context still has a user_state_cache_token from a previous bundle that was not cleared, it raises 'Cache tokens already set to %s', guarding against cache token collisions between bundles.","triggerScenarios":"Entering process_instruction_id twice without the previous contextmanager exiting; exception paths skipping the finally-reset of user_state_cache_token; a runner sending cache tokens for overlapping instructions on the same handler.","commonSituations":"Runner bugs reusing bundle contexts; leaked context bindings after exceptions; shared thread-local state across bundles.","solutions":["Ensure each `with handler.process_instruction_id(...)` completes before the next begins","Restart/retry the bundle to clear leaked thread-local state","Verify only one bundle is executed per harness thread at a time; upgrade Beam if a token-reset bug is present"],"exampleFix":"// before\nwith handler.process_instruction_id(bundle_a, tokens):\n  with handler.process_instruction_id(bundle_b, tokens2):  # raises\n    ...\n// after\nwith handler.process_instruction_id(bundle_a, tokens):\n  ...\nwith handler.process_instruction_id(bundle_b, tokens2):\n  ...","handlingStrategy":"try-catch","validationCode":"if getattr(handler._context, 'user_state_cache_token', None) is not None: raise StateError('cache token still bound')","typeGuard":"def cache_token_bound(handler): return getattr(handler._context, 'user_state_cache_token', None) is not None","tryCatchPattern":"try:\n  with handler.process_instruction_id(bundle_id, cache_tokens):\n    ...\nexcept RuntimeError as e:\n  if str(e).startswith('Cache tokens already set'):\n    log.error('leaked cache token context: %s', e)\n  raise","preventionTips":["Process one bundle at a time per handler","Ensure cache-token context managers always exit (finally)","Check runner for concurrent instruction dispatch bugs"],"tags":["apache-beam","state-handler","cache-tokens","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"}