EmrAddStepsOperator does not work with `deferrable=True`

[cliebBS]

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:

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?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[boring-cyborg]

Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.