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

Code before:
                        waiter=waiter,
                        waiter_max_attempts=self.waiter_max_attempts,
                        waiter_delay=self.waiter_delay,
                        args={"applicationId": self.application_id, "jobRunId": self.job_id},
                        failure_message="Serverless Job failed",
                        status_message="Serverless Job status is",
                        status_args=["jobRun.state", "jobRun.stateDetails"],
                    )
                except AirflowException as e:
                    if "Waiter error: max attempts reached" in str(e):
                        self.log.info(
                            "Cancelling EMR Serverless job %s due to max waiter attempts reached", self.job_id
                        )
                        self.hook.conn.cancel_job_run(applicationId=self.application_id, jobRunId=self.job_id)
                    raise

        return self.job_id

    def execute_complete(self, context: Context, event: dict[str, Any] | None = None) -> None:
        validated_event = validate_execute_complete_event(event)

        if validated_event["status"] == "success":
            self.log.info("Serverless job completed")
            return validated_event["job_id"]
        self.log.info("Cancelling EMR Serverless job %s", self.job_id)
        self.hook.conn.cancel_job_run(applicationId=self.application_id, jobRunId=self.job_id)
        raise AirflowException("EMR Serverless job failed or timed out in deferrable mode")

    def on_kill(self) -> None:
        """
        Cancel the submitted job run.

        Note: this method will not run in deferrable mode.
        """
        if self.job_id:
            self.log.info("Stopping job run with jobId - %s", self.job_id)
            response = self.hook.conn.cancel_job_run(applicationId=self.application_id, jobRunId=self.job_id)
            http_status_code = (
                response.get("ResponseMetadata", {}).get("HTTPStatusCode") if response else None
            )
            if http_status_code is None or http_status_code != 200:
                self.log.error("Unable to request query cancel on EMR Serverless. Exiting")
                return
            self.log.info(
                "Polling EMR Serverless for query with id %s to reach final state",
                self.job_id,
            )
            # This should be replaced with a boto waiter when available.
            waiter(
                get_state_callable=self.hook.conn.get_job_run,
                get_state_args={

Review comment (suggestion):
You should never use instance variable in `execute_complete`. The methods `execute` and `execute_method` are executed with 2 different instances, thus you have no guarantee these variables are set and equal the same across instances. You should only use data from the event. The job ID is available in `validated_event["job_id"]` but you need to modify the trigger to also send the application ID as part of the event

Code after:
                        application_id=self.application_id,
                        job_id=self.job_id,
                        waiter_delay=self.waiter_delay,
                        waiter_max_attempts=self.waiter_max_attempts,
                        aws_conn_id=self.aws_conn_id,
                    ),
                    method_name="execute_complete",
                    timeout=timedelta(seconds=self.waiter_max_attempts * self.waiter_delay),
                )
            else:
                try:
                    waiter = self.hook.get_waiter("serverless_job_completed")
                    wait(
                        waiter=waiter,
                        waiter_max_attempts=self.waiter_max_attempts,
                        waiter_delay=self.waiter_delay,
                        args={"applicationId": self.application_id, "jobRunId": self.job_id},
                        failure_message="Serverless Job failed",
                        status_message="Serverless Job status is",
                        status_args=["jobRun.state", "jobRun.stateDetails"],
                    )
                except AirflowException as e:
                    if "Waiter error: max attempts reached" in str(e):
                        self.log.info(
                            "Cancelling EMR Serverless job %s due to max waiter attempts reached", self.job_id
                        )
                        try:
                            self.hook.conn.cancel_job_run(
                                applicationId=self.application_id, jobRunId=self.job_id
                            )
                        except Exception:
                            self.log.exception(
                                "Failed to cancel EMR Serverless job %s after waiter timeout",
                                self.job_id,
                            )
                    raise

        return self.job_id

    def execute_complete(self, context: Context, event: dict[str, Any] | None = None) -> None:
        validated_event = validate_execute_complete_event(event)

        if validated_event["status"] == "success":
            self.log.info("Serverless job completed")
            return validated_event["job_details"]["job_id"]
        self.log.info("Cancelling EMR Serverless job %s", self.job_id)
        self.hook.conn.cancel_job_run(
            applicationId=validated_event["job_details"]["application_id"],
            jobRunId=validated_event["job_details"]["job_id"],
        )
        raise AirflowException("EMR Serverless job failed or timed out in deferrable mode")