### 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.
timeout=timeout,
                trigger=EmrContainerTrigger(
                    virtual_cluster_id=self.virtual_cluster_id,
                    job_id=self.job_id,
                    aws_conn_id=self.aws_conn_id,
                    waiter_delay=self.poll_interval,
                ),
                method_name="execute_complete",
            )
        if self.wait_for_completion:
            query_status = self.hook.poll_query_status(
                self.job_id,
                max_polling_attempts=self.max_polling_attempts,
                poll_interval=self.poll_interval,
            )
            self.check_failure(query_status)
            if not query_status or query_status in EmrContainerHook.INTERMEDIATE_STATES:
                raise AirflowException(
                    f"Final state of EMR Containers job is {query_status}. "
                    f"Max tries of poll status exceeded, query_execution_id is {self.job_id}."
                )
        return self.job_id
    def check_failure(self, query_status):
        if query_status in EmrContainerHook.FAILURE_STATES:
            error_message = self.hook.get_job_failure_reason(self.job_id)
            raise AirflowException(
                f"EMR Containers job failed. Final state is {query_status}. "
                f"query_execution_id is {self.job_id}. Error: {error_message}"
            )
    def execute_complete(self, context, event=None):