apache/beam · error · ValueError

Already staging

Error message

Already staging %s

What it means

register_job records a staging token so the artifact service can track dependency resolution per job. It raises ValueError if the same staging_token is already registered, preventing duplicate registration and silent overwriting of dependency sets.

Solutions

1. Generate a unique staging token per staging session; do not reuse invocation ids.
2. Check 'staging_token in service._jobs_to_stage' before calling register_job.
3. Catch ValueError and treat it as 'already registered' if idempotent re-staging is intended.

Example fix

// before
service.register_job(token, deps)  # second call with same token
// after
if token not in service._jobs_to_stage:
  service.register_job(token, deps)

if staging_token in service._jobs_to_stage:
    raise RuntimeError(f"{staging_token} already registered")

try:
    service.register_job(staging_token, deps)
except ValueError:
    pass  # idempotent re-registration: token already registered

When it happens

Trigger: Calling register_job (or the stage RPC path that invokes it) twice with the same staging_token — e.g. a client retrying a staging invocation whose first attempt already registered the token, or two jobs sharing an invocation id.

Common situations: Client-side retries after network failures mid-staging; reusing an environment/invocation id across runs; test harnesses invoking register_job manually more than once.

Understand the failure class

Background: "already exists" / EEXIST / FileAlreadyExistsException: what the 'file already exists' error means and how to fix it — this error's family across 37 libraries.

Related errors

• Invalid model update
• key maps to multiple model handlers. All keys must map to…
• NotImplementedError(request.artifact.type_urn)
• A BigQuery table or a query must be specified
• A cluster_identifier should be Optional[Union[str…

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

Appendix: source

Thrown at sdks/python/apache_beam/runners/portability/artifact_service.py:112

    beam_artifact_api_pb2_grpc.ArtifactStagingServiceServicer):
  def __init__(
      self,
      file_writer: Callable[[str, Optional[str]], tuple[BinaryIO, str]],
  ):
    self._lock = threading.Lock()
    self._jobs_to_stage: dict[
        str,
        tuple[dict[Any, list[beam_runner_api_pb2.ArtifactInformation]],
              threading.Event]] = {}
    self._file_writer = file_writer

  def register_job(
      self,
      staging_token: str,
      dependency_sets: MutableMapping[
          Any, list[beam_runner_api_pb2.ArtifactInformation]]):
    if staging_token in self._jobs_to_stage:
      raise ValueError('Already staging %s' % staging_token)
    with self._lock:
      self._jobs_to_stage[staging_token] = (
          dict(dependency_sets), threading.Event())

  def resolved_deps(self, staging_token, timeout=None):
    with self._lock:
      dependency_sets, event = self._jobs_to_stage[staging_token]
    try:
      if not event.wait(timeout):
        raise concurrent.futures.TimeoutError()
      return dependency_sets
    finally:
      with self._lock:
        del self._jobs_to_stage[staging_token]

  def ReverseArtifactRetrievalService(self, responses, context=None):
    staging_token = next(responses).staging_token
    with self._lock:

View on GitHub (pinned to 12126d8942)