{"record":{"id":"413af00561289951","repo":"apache/beam","slug":"job-does-not-exist-flink-uber-jar-server","errorCode":null,"errorMessage":"Job {} does not exist","messagePattern":"Job (.+?) does not exist","errorType":"exception","errorClass":"LookupError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/runners/portability/flink_uber_jar_job_server.py","lineNumber":102,"sourceCode":"  def flink_version(self):\n    full_version = requests.get(\n        '%s/v1/config' % self._master_url, timeout=60).json()['flink-version']\n    # Only return up to minor version.\n    return '.'.join(full_version.split('.')[:2])\n\n  def create_beam_job(self, job_id, job_name, pipeline, options):\n    return FlinkBeamJob(\n        self._master_url,\n        self.executable_jar(),\n        job_id,\n        job_name,\n        pipeline,\n        options,\n        artifact_port=self._artifact_port)\n\n  def GetJobMetrics(self, request, context=None):\n    if request.job_id not in self._jobs:\n      raise LookupError(\"Job {} does not exist\".format(request.job_id))\n    metrics_text = self._jobs[request.job_id].get_metrics()\n    response = beam_job_api_pb2.GetJobMetricsResponse()\n    json_format.Parse(metrics_text, response)\n    return response\n\n\nclass FlinkBeamJob(abstract_job_service.UberJarBeamJob):\n  \"\"\"Runs a single Beam job on Flink by staging all contents into a Jar\n  and uploading it via the Flink Rest API.\"\"\"\n  def __init__(\n      self,\n      master_url,\n      executable_jar,\n      job_id,\n      job_name,\n      pipeline,\n      options,\n      artifact_port=0):","sourceCodeStart":84,"sourceCodeEnd":120,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/runners/portability/flink_uber_jar_job_server.py#L84-L120","documentation":"FlinkUberJarJobServer.GetJobMetrics proxies metrics from the Flink REST API for jobs it launched. It raises LookupError when request.job_id is not in the server's in-memory _jobs dict — this job server process has no record of that job.","triggerScenarios":"Calling the BeamJobApi GetJobMetrics RPC with a job_id this job server did not create in its current process lifetime (server restarted, or id from a job run elsewhere).","commonSituations":"Fetching metrics after the job-server process exited and was relaunched; querying metrics for a job submitted directly to Flink rather than through this Beam job server; copying a job_id from another cluster's logs.","solutions":["Request metrics with a job_id returned by this job server's create_beam_job/run call.","Re-launch the pipeline via the same job server if its process was restarted (in-memory registry lost).","Query the Flink REST API directly (jobs/<job_id>) for jobs not launched through this server."],"exampleFix":"// before\nstub.GetJobMetrics(beam_job_api_pb2.GetJobMetricsRequest(job_id=foreign_id))\n// after\nresult = launcher.run()  # same FlinkUberJarJobServer instance\nstub.GetJobMetrics(beam_job_api_pb2.GetJobMetricsRequest(job_id=result.job_id))","handlingStrategy":"try-catch","validationCode":"# request metrics only for jobs launched by this server instance\nassert job_id in launched_job_ids_on_this_server","typeGuard":null,"tryCatchPattern":"try:\n    metrics = stub.GetJobMetrics(req)\nexcept LookupError:\n    metrics = fetch_metrics_from_flink_rest(job_id)  # fallback","preventionTips":["Only query metrics for jobs launched via the same job server process.","Remember _jobs is in-memory; a server restart invalidates job ids.","Fall back to the Flink REST API for externally submitted jobs."],"tags":["python","apache-beam","flink","grpc-job-service","lookup"],"backgroundTag":"resource-not-found","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}