{"record":{"id":"746a45325ca1f6af","repo":"apache/beam","slug":"the-job-named-s-with-id-s-has-already-been-updated-into-job","errorCode":null,"errorMessage":"The job named %s with id: %s has already been updated into job id: %s and cannot be updated again.","messagePattern":"The job named (.+?) with id: (.+?) has already been updated into job id: (.+?) and cannot be updated again\\.","errorType":"exception","errorClass":"DataflowJobAlreadyExistsError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/runners/dataflow/internal/apiclient.py","lineNumber":914,"sourceCode":"    request.project_id = self.google_cloud_options.project\n    request.location = self.google_cloud_options.region\n    request.job = job.proto\n\n    try:\n      response = self._jobs_client.create_job(request=request)\n    except exceptions.GoogleAPICallError as e:\n      _LOGGER.error(\n          'HTTP status %d trying to create job'\n          ' at dataflow service endpoint %s',\n          e.code,\n          self.google_cloud_options.dataflow_endpoint)\n      _LOGGER.fatal('details of server error: %s', e)\n      raise\n\n    if response.client_request_id and \\\n        response.client_request_id != job.proto.client_request_id:\n      if self.google_cloud_options.update:\n        raise DataflowJobAlreadyExistsError(\n            \"The job named %s with id: %s has already been updated into job \"\n            \"id: %s and cannot be updated again.\" %\n            (response.name, job.proto.replace_job_id, response.id))\n      else:\n        raise DataflowJobAlreadyExistsError(\n            'There is already active job named %s with id: %s. If you want to '\n            'submit a second job, try again by setting a different name using '\n            '--job_name.' % (response.name, response.id))\n\n    _LOGGER.info('Create job: %s', response)\n    # The response is a Job proto with the id for the new job.\n    _LOGGER.info('Created job with id: [%s]', response.id)\n    _LOGGER.info('Submitted job: %s', response.id)\n    _LOGGER.info(\n        'To access the Dataflow monitoring console, please navigate to '\n        'https://console.cloud.google.com/dataflow/jobs/%s/%s?project=%s',\n        self.google_cloud_options.region,\n        response.id,","sourceCodeStart":896,"sourceCodeEnd":932,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/runners/dataflow/internal/apiclient.py#L896-L932","documentation":"DataflowJobAlreadyExistsError thrown during job submission when the service reports the job was already updated into a different job id. This happens only when --update is enabled: the server's client_request_id indicates the update request was already processed, so the same update cannot be replayed. Beam surfaces it so users know their update already succeeded under a new job id.","triggerScenarios":"Submitting a pipeline with --update where the Dataflow service returns a response whose client_request_id differs from the newly generated one — i.e. the job with that name was already updated previously via a retried/submitted request.","commonSituations":"Re-running a pipeline that was already successfully updated; retrying a submission after a network timeout where the first attempt actually succeeded; CI jobs re-running with the same --job_name and --update.","solutions":["Treat the new job id in the message as the active job — monitor that job instead of resubmitting.","Use a fresh job name (or remove --update) if you genuinely want a new job.","Check the Dataflow console/API for the existing job before resubmitting."],"exampleFix":"// before\np.run()  # with --update, same job_name, retried after timeout\n// after\n# detect and reuse\ntry:\n  result = p.run()\nexcept DataflowJobAlreadyExistsError as e:\n  logging.info('Job already updated: %s', e)  # resume with returned job id","handlingStrategy":"try-catch","validationCode":"# before submitting with --update, check whether the named job was already updated\nexisting = [j for j in dataflow_client.list_jobs(project=proj, location=loc).jobs\n            if j.name == job_name and j.current_state in ('JOB_STATE_RUNNING','JOB_STATE_DRAINING')]","typeGuard":null,"tryCatchPattern":"try:\n    result = pipeline.run()\nexcept DataflowJobAlreadyExistsError as e:\n    logging.info('Update already applied: %s', e)  # parse and reuse the new job id\n    return","preventionTips":["Don't blindly retry --update submissions after timeouts; check job state first.","Use idempotent, unique job names per pipeline version.","Inspect the Dataflow console/API before resubmitting."],"tags":["python","apache-beam","dataflow","duplicate-job"],"backgroundTag":"file-already-exists","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"}