### Body

Use `AirflowProviderDeprecationWarning` as warning source in EMR operators and change stacklevel to 2:
https://github.com/apache/airflow/blob/05036e619c0c6dafded1451daac4e07e20aee33f/airflow/providers/amazon/aws/operators/emr.py#L380-L384
https://github.com/apache/airflow/blob/05036e619c0c6dafded1451daac4e07e20aee33f/airflow/providers/amazon/aws/operators/emr.py#L373-L377
https://github.com/apache/airflow/blob/05036e619c0c6dafded1451daac4e07e20aee33f/airflow/providers/amazon/aws/operators/emr.py#L264-L268
https://github.com/apache/airflow/blob/05036e619c0c6dafded1451daac4e07e20aee33f/airflow/providers/amazon/aws/operators/emr.py#L257-L261

### Committer

- [X] I acknowledge that I am a maintainer/committer of the Apache Airflow project.
if event["status"] != "success":
            raise AirflowException(f"Error while running job: {event}")
        self.log.info("%s", event["message"])
        return event["job_id"]
    def on_kill(self) -> None:
        """Cancel the submitted job run."""
        if self.job_id:
            self.log.info("Stopping job run with jobId - %s", self.job_id)
            response = self.hook.stop_query(self.job_id)
            http_status_code = None
            try:
                http_status_code = response["ResponseMetadata"]["HTTPStatusCode"]
            except Exception as ex:
                self.log.error("Exception while cancelling query: %s", ex)
            finally:
                if http_status_code is None or http_status_code != 200:
                    self.log.error("Unable to request query cancel on EMR. Exiting")
                else:
                    self.log.info(
                        "Polling EMR for query with id %s to reach final state",
                        self.job_id,
                    )
                    self.hook.poll_query_status(self.job_id)
class EmrCreateJobFlowOperator(BaseOperator):