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

Code before:
        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=""),
            job_run_id=self._job_run_id,
        )
        self.log.info("You can monitor this Glue Job run at: %s", glue_job_run_url)

        if self.deferrable:
            self.defer(
                trigger=GlueJobCompleteTrigger(
                    job_name=self.job_name,
                    run_id=self._job_run_id,
                    verbose=self.verbose,
                    aws_conn_id=self.aws_conn_id,
                    waiter_delay=self.waiter_delay,

Review comment (suggestion):
This seems overly defensive to me; `context["ti"]` should always be available.

Code after:
        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)

        if self.deferrable:
            self.defer(
                trigger=GlueJobCompleteTrigger(
                    job_name=self.job_name,
                    run_id=self._job_run_id,
                    verbose=self.verbose,
                    aws_conn_id=self.aws_conn_id,
                    waiter_delay=self.waiter_delay,
                    waiter_max_attempts=self.waiter_max_attempts,
                    region_name=self.region_name,
                ),