Fix automatic termination issue in EmrOperator by ensuring waiter_max_attempts is set for deferrable triggers

[beobest2]

When running EMR jobs that require a long duration, setting the waiter_max_attempts in EmrOperator to a high value does not prevent the "waiter error: max attempts" error from occurring after a certain period.

[2024-03-27 16:42:08,646] Trigger failed:
Traceback (most recent call last):

  File "/usr/local/airflow/.local/lib/python3.11/site-packages/airflow/jobs/triggerer_job_runner.py", line 529, in cleanup_finished_triggers
    result = details["task"].result()
             ^^^^^^^^^^^^^^^^^^^^^^^^

  File "/usr/local/airflow/.local/lib/python3.11/site-packages/airflow/jobs/triggerer_job_runner.py", line 601, in run_trigger
    async for event in trigger.run():

  File "/usr/local/airflow/.local/lib/python3.11/site-packages/airflow/providers/amazon/aws/triggers/base.py", line 143, in run
    await async_wait(

  File "/usr/local/airflow/.local/lib/python3.11/site-packages/airflow/providers/amazon/aws/utils/waiter_with_logging.py", line 133, in async_wait
    raise AirflowException("Waiter error: max attempts reached")

---

An issue has been identified with the AWS provider code.
The EmrOperator that calls this trigger does not allow users to input waiter_max_attempts, resulting in the application of the default value of 600.

airflow/providers/amazon/aws/triggers/emr.py

class EmrContainerTrigger(AwsBaseWaiterTrigger):
    """
    Poll for the status of EMR container until reaches terminal state.
    :param virtual_cluster_id: Reference Emr cluster id
    :param job_id:  job_id to check the state
    :param aws_conn_id: Reference to AWS connection id
    :param waiter_delay: polling period in seconds to check for the status
    """

    def __init__(
        self,
        virtual_cluster_id: str,
        job_id: str,
        aws_conn_id: str | None = "aws_default",
        poll_interval: int | None = None,  # deprecated
        waiter_delay: int = 30,
        waiter_max_attempts: int = 600,   # <- We can't modify this ..
    ):

In other words, if the poll_interval is set to 60 seconds, the job will terminate after 10 hours (1 minute * 600) regardless of the waiter_max_attempts value specified by the user.

To address this, modifications were made in EmrOperator to ensure that parameters provided by the user are also passed to the EmrContainerTrigger.

[boring-cyborg]

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contributors' Guide (https://github.com/apache/airflow/blob/main/contributing-docs/README.rst)
Here are some useful points:

• Pay attention to the quality of your code (ruff, mypy and type annotations). Our pre-commits will help you with that.
• In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example DAG that shows how users should use it.
• Consider using Breeze environment for testing locally, it's a heavy docker but it ships with a working Airflow and a lot of integrations.
• Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
• Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
• Be sure to read the Airflow Coding style.
• Always keep your Pull Requests rebased, otherwise your build might fail due to changes not related to your commits.
  Apache Airflow is a community-driven project and together we are making it better 🚀.
  In case of doubts contact the developers at:
  Mailing List: dev@airflow.apache.org
  Slack: https://s.apache.org/airflow-slack

[matcha-1atte]

👍

[matcha-1atte]

Good.

[eladkal]

can you add unit tests to cover this change?

[beobest2]

@eladkal I have also added unit tests. Please review

[o-nikolas]

Tests need to pass after your rebase from main. Otherwise looks good to merge

[o-nikolas]

Running workflows again after recent commits

[beobest2]

@o-nikolas Previously, I was unable to properly test in my development environment because different dependency errors occurred. After solving the problem and running tests, I found that when deferrable is set to True, the call_count is not recorded as many times as the actual number of attempts we specified. Do you have an idea how can we verify the call_count in a unit test when using deferrable as it is now?

[o-nikolas]

@o-nikolas Previously, I was unable to properly test in my development environment because different dependency errors occurred. After solving the problem and running tests, I found that when deferrable is set to True, the call_count is not recorded as many times as the actual number of attempts we specified. Do you have an idea how can we verify the call_count in a unit test when using deferrable as it is now?

CC @syedahsn

[syedahsn]

Left a couple of comments. Thanks for looking into this!

[syedahsn]

LGTM

[boring-cyborg]

Awesome work, congrats on your first merged pull request! You are invited to check our Issue Tracker for additional contributions.