apache/beam · error · RuntimeError

The requesting owner must be registered.

Error message

The requesting owner must be registered.

What it means

The _SharedCache.get() method refuses to return a cached resource (like a subprocess server handle) when the caller supplies an owner name that is not among the currently registered live owners. Owners are registered when they first use the cache and unregistered when they release the resource; this check prevents an unregistered (already released) owner from re-acquiring or keeping shared resources alive. It is a bookkeeping/state error, not a resource failure.

Source

Thrown at sdks/python/apache_beam/utils/subprocess_server.py:116

            owner)
        return
      del self._live_owners[owner]
      for key, entry in list(self._cache.items()):
        if owner in entry.owners:
          entry.owners.remove(owner)
        if not entry.owners:
          to_delete.append(entry.obj)
          del self._cache[key]
    # Actually call the destructors outside of the lock.
    for value in to_delete:
      self._destructor(value)

  def get(self, *key, owner=None):
    with self._lock:
      if not self._live_owners:
        raise RuntimeError("At least one owner must be registered.")
      if owner is not None and owner not in self._live_owners:
        raise RuntimeError("The requesting owner must be registered.")

      if key not in self._cache:
        self._cache[key] = _SharedCacheEntry(self._constructor(*key), set())
      if owner is not None:
        self._cache[key].owners.add(owner)
        for live_owner, is_context in self._live_owners.items():
          if is_context:
            self._cache[key].owners.add(live_owner)
      else:
        for live_owner in self._live_owners:
          self._cache[key].owners.add(live_owner)
      return self._cache[key].obj

  def force_remove(self, *key):
    with self._lock:
      entry = self._cache.pop(key, None)
    if entry is not None:
      self._destructor(entry.obj)

View on GitHub (pinned to 12126d8942)

Solutions

1. Ensure the owner that calls get() is the same object that originally registered (typically the 'with' context owner).
2. Re-enter the owning context manager so the owner is re-registered before calling get().
3. Do not pass an owner at all if you only need to look up an existing cache entry.
4. Fix owner lifecycle ordering so all consumers release owners only after the last get().

Example fix

// before
shared_cache.get(('jar',), owner=released_owner)
// after
with shared_cache_entries(...) as owner:
    shared_cache.get(('jar',), owner=owner)

def owner_is_registered(cache, owner):
    return owner is None or owner in cache._live_owners
# call get() only if owner_is_registered(cache, owner)

def is_live_owner(cache, owner):
    return owner is None or owner in cache._live_owners

When it happens

Trigger: Calling get(key..., owner=X) after X was removed from _live_owners (e.g. after a context manager exit or release), or passing an owner that was never registered, while at least one other owner is live.

Common situations: Reusing a cache/context-manager object after its 'with' block ended; nested pipelines sharing a subprocess server where one owner released it before another tries to access it; passing the wrong owner string/handle.

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

• {module} is not registered for pickle by value
• PCollection not available, please run the pipeline.
• input is already set.
• Merging requested for non-mergeable state tag: %r.
• Invalid tag.

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