apache/beam · error · LookupError

Job {} does not exist

Error message

Job {} does not exist

What it means

LocalJobService.GetJobMetrics looks up the requested job_id in its _jobs registry and raises LookupError if the id is unknown. This is a client-facing gRPC servicer method, so requesting metrics for a job the local server never started (or has since discarded) produces this error.

Source

Thrown at sdks/python/apache_beam/runners/portability/local_job_service.py:156

        '%s:%d' % (self.get_bind_address(), port))
    beam_job_api_pb2_grpc.add_JobServiceServicer_to_server(self, self._server)
    beam_artifact_api_pb2_grpc.add_ArtifactStagingServiceServicer_to_server(
        self._artifact_service, self._server)
    hostname = self.get_service_address()
    self._artifact_staging_endpoint = endpoints_pb2.ApiServiceDescriptor(
        url='%s:%d' % (hostname, port))
    self._server.start()
    _LOGGER.info('Grpc server started at %s on port %d' % (hostname, port))
    return port

  def stop(self, timeout=1):
    self._server.stop(timeout)
    if os.path.exists(self._staging_dir) and self._cleanup_staging_dir:
      shutil.rmtree(self._staging_dir, ignore_errors=True)

  def GetJobMetrics(self, request, context=None):
    if request.job_id not in self._jobs:
      raise LookupError("Job {} does not exist".format(request.job_id))

    result = self._jobs[request.job_id].result
    if result is None:
      monitoring_info_list = []
    else:
      monitoring_info_list = result.monitoring_infos()

    # Filter out system metrics
    user_monitoring_info_list = [
        x for x in monitoring_info_list
        if monitoring_infos.is_user_monitoring_info(x)
    ]

    return beam_job_api_pb2.GetJobMetricsResponse(
        metrics=beam_job_api_pb2.MetricResults(
            committed=user_monitoring_info_list))

View on GitHub (pinned to 12126d8942)

Solutions

1. Use the job_id returned by the SubmitJob/RunJob response for this server instance.
2. Verify the local job service process is still the same one that ran the job (no restart).
3. List/track jobs you submitted (e.g. keep the result handle) rather than reconstructing ids.
4. Re-run the pipeline if its job record was lost on restart, then query metrics from the new id.

Example fix

// before
service.GetJobMetrics(request(job_id='job_0042'))  # guessed id
// after
result, job_id = service.RunJob(...)  # keep returned job_id
service.GetJobMetrics(request(job_id=job_id))

if not job_id or job_id not in known_job_ids:
    raise SystemExit('unknown job_id: %s' % job_id)

def job_known(service, job_id): return job_id in service._jobs

try:
    metrics = service.GetJobMetrics(req)
except LookupError:
    log.warning('job %s not found on this server', req.job_id)

When it happens

Trigger: Calling GetJobMetrics with a job_id that was never submitted to this local_job_service instance, a mistyped id, or an id from a previous server process that has since restarted.

Common situations: Sharing job ids across separate local runner invocations; server restarted between job submission and metrics query; copy/paste typos in --job_id; querying a job served by a different BeamFnApiRunner instance.

Understand the failure class

Background: "Not found" and "does not exist" errors: why "Task not found", "No such folder", and "Can't find" fire when a lookup comes back empty — this error's family across 14 libraries.

Related errors

• State stream is closed.
• Unknown type
• Instruction id was registered twice
• Logging stream terminated unexpectedly with success before i
• All workers communicate through gRPC should have worker_id.

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