### Apache Airflow version

2.7.0

### What happened

When using the `EmrAddStepsOperator`, trying to use `deferrable=True` causes an exception from the Amazon provider:

```
[2023-09-07, 16:02:56 UTC] {taskinstance.py:1720} ERROR - Trigger failed:
Traceback (most recent call last):
  File "/home/***/.local/lib/python3.11/site-packages/***/jobs/triggerer_job_runner.py", line 527, in cleanup_finished_triggers
    result = details["task"].result()
             ^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/***/.local/lib/python3.11/site-packages/***/jobs/triggerer_job_runner.py", line 599, in run_trigger
    async for event in trigger.run():
  File "/home/***/.local/lib/python3.11/site-packages/***/providers/amazon/aws/triggers/emr.py", line 74, in run
    for attempt in range(1, 1 + self.max_attempts):
                            ~~^~~~~~~~~~~~~~~~~~~
TypeError: unsupported operand type(s) for +: 'int' and 'str'
```

The problem seems to stem from the `max_attempts` value being stored as an `int` when the `EmrAddStepsTrigger` class is instantiated, but when `run` is called later, it becomes a `str`.  The simple fix is to wrap the access with `int()`, but I'm not sure if that's the _correct_ fix.  For some reason, `EmrAddStepsTrigger` extends `BaseTrigger` from Airflow instead of `AwsBaseWaiterTrigger` like the rest of the EMR triggers even though it uses an AWS waiter.  Perhaps the more appropriate fix would be to rewrite this trigger to use the AWS-specific base class.

### What you think should happen instead

The `EmrAddStepOperator` should work when deferred

### How to reproduce

Minimal DAG, run using the `apache/airflow:2.7.0-python3.11` image with the provided docker-compose.yaml file:

```python
from datetime import datetime

from airflow import DAG
from airflow.providers.amazon.aws.operators.emr import EmrAddStepsOperator

with DAG(
    "emr_defer_test",
    start_date=datetime(2021, 1, 1),
    catchup=False,
    schedule_interval=None,
) as dag:
    # this will work on any EMR cluster, put your cluster ID here
    job_flow_id = "j-1M0574ZKPRD9H"

    EmrAddStepsOperator(
        task_id="run_deferred",
        job_flow_id=job_flow_id,
        steps=[
            {
                "Name": "hello-world",
                "ActionOnFailure": "CONTINUE",
                "HadoopJarStep": {
                    "Jar": "command-runner.jar",
                    "Args": "sleep 60".split(),
                }
            }
        ],
        deferrable=True,
    )
```

### Operating System

Docker on MacOS 13.5.1 via colima

### Versions of Apache Airflow Providers

```
apache-airflow-providers-amazon==8.5.1
apache-airflow-providers-celery==3.3.2
apache-airflow-providers-common-sql==1.7.0
apache-airflow-providers-ftp==3.5.0
apache-airflow-providers-http==4.5.0
apache-airflow-providers-imap==3.3.0
apache-airflow-providers-postgres==5.6.0
apache-airflow-providers-sqlite==3.4.3
```

### Deployment

Docker-Compose

### 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)

def test_service_waiters(self):
        hook = EksHook()
        with open(hook.waiter_path) as config_file:
            expected_waiters = json.load(config_file)["waiters"]
        for waiter in list(expected_waiters.keys()):
            assert waiter in hook.list_waiters()
            assert waiter in hook._list_custom_waiters()
    @mock_eks
    def test_existing_waiter_inherited(self):
        """
        AwsBaseHook::get_waiter will first check if there is a custom waiter with the
        provided name and pass that through is it exists, otherwise it will check the
        custom waiters for the given service.  This test checks to make sure that the
        waiter is the same whichever way you get it and no modifications are made.
        """
        hook_waiter = EksHook().get_waiter("cluster_active")
        client_waiter = EksHook().conn.get_waiter("cluster_active")
        boto_waiter = boto3.client("eks").get_waiter("cluster_active")
        assert_all_match(hook_waiter.name, client_waiter.name, boto_waiter.name)
        assert_all_match(len(hook_waiter.__dict__), len(client_waiter.__dict__), len(boto_waiter.__dict__))
        for attr in hook_waiter.__dict__:
            
            
            assert hasattr(boto_waiter, attr)
            assert hasattr(client_waiter, attr)
class TestCustomECSServiceWaiters: