apache/beam · error · RuntimeError

Cache tokens already set to

Error message

Cache tokens already set to %s

What it means

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.

Solutions

  1. Ensure each `with handler.process_instruction_id(...)` completes before the next begins
  2. Restart/retry the bundle to clear leaked thread-local state
  3. Verify only one bundle is executed per harness thread at a time; upgrade Beam if a token-reset bug is present

Example fix

// before
with handler.process_instruction_id(bundle_a, tokens):
  with handler.process_instruction_id(bundle_b, tokens2):  # raises
    ...
// after
with handler.process_instruction_id(bundle_a, tokens):
  ...
with handler.process_instruction_id(bundle_b, tokens2):
  ...
Defensive patterns

Strategy: try-catch

Validate before calling

if getattr(handler._context, 'user_state_cache_token', None) is not None: raise StateError('cache token still bound')

Type guard

def cache_token_bound(handler): return getattr(handler._context, 'user_state_cache_token', None) is not None

Try / catch

try:
  with handler.process_instruction_id(bundle_id, cache_tokens):
    ...
except RuntimeError as e:
  if str(e).startswith('Cache tokens already set'):
    log.error('leaked cache token context: %s', e)
  raise

Prevention

When it happens

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

Common situations: Runner bugs reusing bundle contexts; leaked context bindings after exceptions; shared thread-local state across bundles.

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

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

Appendix: source

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

      self,
      global_state_cache,  # type: StateCache
      underlying_state  # type: StateHandler
  ):
    # type: (...) -> None
    self._underlying = underlying_state
    self._state_cache = global_state_cache
    self._context = threading.local()

    # state retrieval time statistics
    self._retrieval_time = 0.0
    self._get_raw_called = 0
    self._warn_interval = 60.0

  @contextlib.contextmanager
  def process_instruction_id(self, bundle_id, cache_tokens):
    # type: (str, Iterable[beam_fn_api_pb2.ProcessBundleRequest.CacheToken]) -> Iterator[None]
    if getattr(self._context, 'user_state_cache_token', None) is not None:
      raise RuntimeError(
          'Cache tokens already set to %s' %
          self._context.user_state_cache_token)
    self._context.side_input_cache_tokens = {}
    user_state_cache_token = None
    for cache_token_struct in cache_tokens:
      if cache_token_struct.HasField("user_state"):
        # There should only be one user state token present
        assert not user_state_cache_token
        user_state_cache_token = cache_token_struct.token
      elif cache_token_struct.HasField("side_input"):
        self._context.side_input_cache_tokens[
            cache_token_struct.side_input.transform_id,
            cache_token_struct.side_input.
            side_input_id] = cache_token_struct.token
    # TODO: Consider a two-level cache to avoid extra logic and locking
    # for items cached at the bundle level.
    self._context.bundle_cache_token = bundle_id
    try:

View on GitHub (pinned to 12126d8942)