apache/beam · error · RuntimeError

At least one owner must be registered.

Error message

At least one owner must be registered.

What it means

_SharedCache in apache_beam/utils/subprocess_server.py caches shared resources (e.g. subprocess servers) keyed by owner. get() requires at least one owner to have been registered via register_owner; with an empty owner set the cache cannot safely create/return entries, so it raises this RuntimeError.

Source

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

            "Subprocess owner %s already purged. If this occurs during atexit "
            "shutdown, the subprocess was already cleaned up earlier.",
            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)

View on GitHub (pinned to 12126d8942)

Solutions

  1. Register an owner first via cache.register_owner(owner) before calling get().
  2. Pass owner=<registered owner> to get() so usage is tracked for cleanup.
  3. Ensure teardown/unregister isn't called before the final get() in your lifecycle.
  4. If driving Beam internals directly, create the cache within the standard Pipeline machinery so owners are registered automatically.

Example fix

// before
server = shared_cache.get('beam', jar)  # RuntimeError
// after
shared_cache.register_owner('pipeline-1')
server = shared_cache.get('beam', jar, owner='pipeline-1')
Defensive patterns

Strategy: validation

Validate before calling

if not shared_cache._live_owners:
    raise RuntimeError('register an owner before using the shared cache')

Type guard

def cache_is_ready(cache) -> bool:
    return bool(getattr(cache, '_live_owners', None))

Try / catch

try:
    server = cache.get(key, owner=owner)
except RuntimeError as e:
    logging.error('Shared cache misuse: %s', e)
    cache.register_owner(owner)
    server = cache.get(key, owner=owner)

Prevention

When it happens

Trigger: Calling cache.get(*key) (with or without owner=...) before any register_owner() call, or after all owners were unregistered (e.g. all pipelines closed and cleanup ran).

Common situations: Invoking subprocess server helpers outside the normal Beam pipeline lifecycle; calling get() after service shutdown/teardown; custom code constructing a _SharedCache and forgetting to register an owner.

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/5cc70414481e3e6a. Report an issue: GitHub.