Repository: apache/airflow
File: providers/amazon/src/airflow/providers/amazon/aws/operators/glue.py (Python)

Code before:
            self.s3_artifacts_prefix + script_name,
            bucket_name=self.s3_bucket,
            replace=self.replace_script_file,
        )
        self.s3_script_location = f"s3://{self.s3_bucket}/{self.s3_artifacts_prefix}{script_name}"

    def execute(self, context: Context):
        """
        Execute AWS Glue Job from Airflow.

        :return: the current Glue job ID.
        """
        previous_job_run_id = None
        if self.resume_glue_job_on_retry:
            ti = context.get("ti")
            if ti:
                previous_job_run_id = ti.xcom_pull(key="glue_job_run_id", task_ids=ti.task_id)
                if previous_job_run_id:
                    try:
                        job_run = self.hook.conn.get_job_run(JobName=self.job_name, RunId=previous_job_run_id)
                        state = job_run.get("JobRun", {}).get("JobRunState")
                        self.log.info("Previous Glue job_run_id: %s, state: %s", previous_job_run_id, state)
                        if state in ("RUNNING", "STARTING", "STOPPING"):
                            self._job_run_id = previous_job_run_id
                    except Exception as e:
                        self.log.warning("Failed to get previous Glue job run state: %s", e)

        if not self._job_run_id:
            self.log.info(
                "Initializing AWS Glue Job: %s. Wait for completion: %s",
                self.job_name,
                self.wait_for_completion,
            )
            glue_job_run = self.hook.initialize_job(self.script_args, self.run_job_kwargs)
            self._job_run_id = glue_job_run["JobRunId"]
            ti = context.get("ti")
            if ti:
                ti.xcom_push(key="glue_job_run_id", value=self._job_run_id)

        glue_job_run_url = GlueJobRunDetailsLink.format_str.format(
            aws_domain=GlueJobRunDetailsLink.get_aws_domain(self.hook.conn_partition),
            region_name=self.hook.conn_region_name,
            job_name=urllib.parse.quote(self.job_name, safe=""),
            job_run_id=self._job_run_id,
        )
        GlueJobRunDetailsLink.persist(
            context=context,
            operator=self,
            region_name=self.hook.conn_region_name,
            aws_partition=self.hook.conn_partition,
            job_name=urllib.parse.quote(self.job_name, safe=""),

Review comment (bug):
```suggestion
                    except Exception:
                        self.log.warning("Failed to get previous Glue job run state", exc_info=True)
```

Code after:
            self.s3_artifacts_prefix + script_name,
            bucket_name=self.s3_bucket,
            replace=self.replace_script_file,
        )
        self.s3_script_location = f"s3://{self.s3_bucket}/{self.s3_artifacts_prefix}{script_name}"

    def execute(self, context: Context):
        """
        Execute AWS Glue Job from Airflow.

        :return: the current Glue job ID.
        """
        previous_job_run_id = None
        if self.resume_glue_job_on_retry:
            ti = context["ti"]
            previous_job_run_id = ti.xcom_pull(key="glue_job_run_id", task_ids=ti.task_id)
            if previous_job_run_id:
                try:
                    job_run = self.hook.conn.get_job_run(JobName=self.job_name, RunId=previous_job_run_id)
                    state = job_run.get("JobRun", {}).get("JobRunState")
                    self.log.info("Previous Glue job_run_id: %s, state: %s", previous_job_run_id, state)
                    if state in ("RUNNING", "STARTING", "STOPPING"):
                        self._job_run_id = previous_job_run_id
                except Exception:
                    self.log.warning("Failed to get previous Glue job run state", exc_info=True)

        if not self._job_run_id:
            self.log.info(
                "Initializing AWS Glue Job: %s. Wait for completion: %s",
                self.job_name,
                self.wait_for_completion,
            )
            glue_job_run = self.hook.initialize_job(self.script_args, self.run_job_kwargs)
            self._job_run_id = glue_job_run["JobRunId"]
            context["ti"].xcom_push(key="glue_job_run_id", value=self._job_run_id)

        glue_job_run_url = GlueJobRunDetailsLink.format_str.format(
            aws_domain=GlueJobRunDetailsLink.get_aws_domain(self.hook.conn_partition),
            region_name=self.hook.conn_region_name,
            job_name=urllib.parse.quote(self.job_name, safe=""),
            job_run_id=self._job_run_id,
        )
        GlueJobRunDetailsLink.persist(
            context=context,
            operator=self,
            region_name=self.hook.conn_region_name,
            aws_partition=self.hook.conn_partition,
            job_name=urllib.parse.quote(self.job_name, safe=""),
            job_run_id=self._job_run_id,
        )
        self.log.info("You can monitor this Glue Job run at: %s", glue_job_run_url)