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

Code before:
                    self.conn.cancel_job_run(applicationId=application_id, jobRunId=job_id)
        if wait_for_completion:
            if count > 0:
                self.log.info("now waiting for the %s cancelled job(s) to terminate", count)
                self.get_waiter("no_job_running").wait(
                    applicationId=application_id,
                    states=list(self.JOB_INTERMEDIATE_STATES.union({"CANCELLING"})),
                    WaiterConfig=waiter_config or {},
                )

        return count

    def cancel_job(self, application_id: str, job_run_id: str) -> None:
        """
        Cancel a specific EMR Serverless job run.

        :param application_id: The ID of the EMR Serverless application
        :param job_run_id: The ID of the job run to cancel
        :raises AirflowException: If cancellation fails
        """
        try:
            self.log.info("Cancelling EMR Serverless job %s in application %s", job_run_id, application_id)
            response = self.conn.cancel_job_run(applicationId=application_id, jobRunId=job_run_id)
            http_code = response["ResponseMetadata"]["HTTPStatusCode"]
            if http_code != 200:
                raise AirflowException(f"Failed to cancel job run {job_run_id}: HTTP {http_code}")
            self.log.info("Cancelled EMR Serverless job run %s successfully", job_run_id)
        except ClientError as e:
            raise AirflowException(f"Error cancelling EMR Serverless job run {job_run_id}: {e}")


def is_connection_being_updated_exception(exception: BaseException) -> bool:
    return (
        isinstance(exception, ClientError)
        and exception.response["Error"]["Code"] == "ValidationException"
        and "is not reachable as its connection is currently being updated"
        in exception.response["Error"]["Message"]
    )


class EmrContainerHook(AwsBaseHook):
    """
    Interact with Amazon EMR Containers (Amazon EMR on EKS).

    Provide thick wrapper around :py:class:`boto3.client("emr-containers") <EMRContainers.Client>`.

    :param virtual_cluster_id: Cluster ID of the EMR on EKS virtual cluster

    Additional arguments (such as ``aws_conn_id``) may be specified and
    are passed down to the underlying AwsBaseHook.


Review comment (suggestion):
As far as I know this can never happen. If the call is not successful, you'll get an exception.

Other than that, this method is just a thin wrapper on top of `self.conn.cancel_job_run`, I dont think it is necessary to create a method for that. You can call boto3 API directly in the operator.

Code after:
                    self.conn.cancel_job_run(applicationId=application_id, jobRunId=job_id)
        if wait_for_completion:
            if count > 0:
                self.log.info("now waiting for the %s cancelled job(s) to terminate", count)
                self.get_waiter("no_job_running").wait(
                    applicationId=application_id,
                    states=list(self.JOB_INTERMEDIATE_STATES.union({"CANCELLING"})),
                    WaiterConfig=waiter_config or {},
                )

        return count


def is_connection_being_updated_exception(exception: BaseException) -> bool:
    return (
        isinstance(exception, ClientError)
        and exception.response["Error"]["Code"] == "ValidationException"
        and "is not reachable as its connection is currently being updated"
        in exception.response["Error"]["Message"]
    )


class EmrContainerHook(AwsBaseHook):
    """
    Interact with Amazon EMR Containers (Amazon EMR on EKS).

    Provide thick wrapper around :py:class:`boto3.client("emr-containers") <EMRContainers.Client>`.

    :param virtual_cluster_id: Cluster ID of the EMR on EKS virtual cluster

    Additional arguments (such as ``aws_conn_id``) may be specified and
    are passed down to the underlying AwsBaseHook.

    .. seealso::
        - :class:`airflow.providers.amazon.aws.hooks.base_aws.AwsBaseHook`
    """

    INTERMEDIATE_STATES = (
        "PENDING",
        "SUBMITTED",
        "RUNNING",
    )
    FAILURE_STATES = (
        "FAILED",
        "CANCELLED",
        "CANCEL_PENDING",
    )
    SUCCESS_STATES = ("COMPLETED",)
    TERMINAL_STATES = (
        "COMPLETED",
        "FAILED",