{"record":{"id":"a76ec710ed3f3ec4","repo":"apache/beam","slug":"found-no-metrics-container-for-job","errorCode":null,"errorMessage":"Found no metrics container for job {}","messagePattern":"Found no metrics container for job (.+?)","errorType":"exception","errorClass":"LookupError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/runners/portability/flink_uber_jar_job_server.py","lineNumber":252,"sourceCode":"        for ix, exc in enumerate(response['all-exceptions']):\n          yield beam_job_api_pb2.JobMessage(\n              message_id='message%d' % ix,\n              time=str(exc['timestamp']),\n              importance=beam_job_api_pb2.JobMessage.MessageImportance.\n              JOB_MESSAGE_ERROR,\n              message_text=exc['exception'])\n        yield state, timestamp\n        break\n      else:\n        yield state, timestamp\n\n  def get_metrics(self):\n    accumulators = self.get('v1/jobs/%s/accumulators' %\n                            self._flink_job_id)['user-task-accumulators']\n    for accumulator in accumulators:\n      if accumulator['name'] == '__metricscontainers':\n        return accumulator['value']\n    raise LookupError(\n        \"Found no metrics container for job {}\".format(self._flink_job_id))\n","sourceCodeStart":234,"sourceCodeEnd":254,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/runners/portability/flink_uber_jar_job_server.py#L234-L254","documentation":"FlinkJobServerBeamJob.get_metrics() queries the Flink REST API for the job's user-task accumulators and looks for one named '__metricscontainers', which Beam pipelines use to carry metrics back from Flink workers. If no accumulator with that name exists, it means the job's tasks never published a metrics container (e.g. the job finished without running any Beam tasks successfully, or metrics were not reported), so a LookupError is raised.","triggerScenarios":"Calling get_metrics() on a FlinkJobServerBeamJob whose Flink job returned no 'user-task-accumulators' entry named '__metricscontainers' — e.g. querying metrics for a job that failed at startup, a job whose workers died before reporting, or a job id that resolved to a job with no Beam task accumulators.","commonSituations":"Polling metrics for an already-finished/failed Flink job; a Flink cluster whose classpath lacks the Beam Flink runner jars so accumulators are never published; querying too early before any task has executed; stale job ids after a job restart.","solutions":["Wait until the job is running and at least one task has executed before calling get_metrics().","Verify the Flink job is healthy via the Flink REST endpoint (v1/jobs/<id>) and that it did not fail before reporting accumulators.","Ensure the Beam Flink runner jar on the job server matches the SDK version so '__metricscontainers' accumulators are emitted.","Wrap get_metrics() in try/except LookupError and return empty/unknown metrics if the container is absent."],"exampleFix":"# before\nmetrics = job.get_metrics()\n# after\ntry:\n  metrics = job.get_metrics()\nexcept LookupError:\n  metrics = None  # job has not reported any metrics container yet","handlingStrategy":"try-catch","validationCode":"job_info = job.get('v1/jobs/%s' % job._flink_job_id)\nif job_info.get('state') not in ('RUNNING', 'FINISHED'):\n    raise RuntimeError('job not running; metrics may be unavailable')","typeGuard":"def has_metrics_container(job):\n    accs = job.get('v1/jobs/%s/accumulators' % job._flink_job_id).get('user-task-accumulators') or []\n    return any(a.get('name') == '__metricscontainers' for a in accs)","tryCatchPattern":"try:\n    metrics = job.get_metrics()\nexcept LookupError:\n    metrics = None  # or retry after the job has run longer","preventionTips":["Only query metrics once the Flink job has at least one running/finished task.","Verify the job server and SDK versions match so metrics accumulators are published.","Check the Flink REST /jobs/<id> endpoint for FAILED/CANCELED state before polling metrics."],"tags":["apache-beam","flink","metrics","portability-runner"],"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-20T03:17:13.778Z"}