### Apache Airflow version

2.5.1

### What happened

When using EMRHook.get_cluster_id_by_name or any any operator that depends on it (e.g. EMRAddStepsOperator), if the results of the ListClusters API call is paginated (e.g. if your account has more than 50 clusters in the current region), and the desired cluster is in the 2nd page of results, None will be returned instead of the cluster ID.

### What you think should happen instead

Boto's pagination API should be used and the cluster ID should be returned.

### How to reproduce

Use `EmrAddStepsOperator` with the `job_flow_name` parameter on an `aws_conn_id` with more than 50 EMR clusters in the current region.

### Operating System

Linux

### Versions of Apache Airflow Providers

apache-airflow-providers-amazon==7.2.1

### Deployment

Official Apache Airflow Helm Chart

### Deployment details

_No response_

### Anything else

_No response_

### Are you willing to submit PR?

- [X] Yes I am willing to submit a PR!

### Code of Conduct

- [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)

:param job_driver: Job configuration details, e.g. the Spark job parameters.
        :param configuration_overrides: The configuration overrides for the job run,
            specifically either application configuration or monitoring configuration.
        :param client_request_token: The client idempotency token of the job run request.
            Use this if you want to specify a unique ID to prevent two jobs from getting started.
        :param tags: The tags assigned to job runs.
        :return: The ID of the job run request.
        """
        params = {
            "name": name,
            "virtualClusterId": self.virtual_cluster_id,
            "executionRoleArn": execution_role_arn,
            "releaseLabel": release_label,
            "jobDriver": job_driver,
            "configurationOverrides": configuration_overrides or {},
            "tags": tags or {},
        }
        if client_request_token:
            params["clientToken"] = client_request_token
        response = self.conn.start_job_run(**params)
        if response["ResponseMetadata"]["HTTPStatusCode"] != 200:
            raise AirflowException(f"Start Job Run failed: {response}")
        else:
            self.log.info(
                "Start Job Run success - Job Id %s and virtual cluster id %s",
                response["id"],
                response["virtualClusterId"],
            )
            return response["id"]
    def get_job_failure_reason(self, job_id: str) -> str | None: