Allow waiter to be configured via EmrServerless Operators

[syedahsn]

Currently, the waiter function used in the wait_for_completion logic for EMR Serverless operators is not configurable. There have been user requests to allow the operators to pass parameters to the waiter. This PR adds this feature. In addition, the waiter function is generic enough to be used outside of the EmrServerlessHook, so it is being moved to a generic location.

closes: aws-samples/emr-serverless-samples#27

---

^ Add meaningful description above

Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in newsfragments.

[syedahsn]

@eladkal can you take another look at this when you have some time? I think I've addressed the issues you brought up. Thanks!

[potiuk]

LGTM. @eladkal ?

[eladkal]

LGTM

[eladkal]

Wonder about

EmrServerlessCreateApplicationOperator(
...,
wait_for_completion=False,
waiter_countdown=40,
waiter_check_interval_seconds=70
)

In this case the waiter_countdown, waiter_check_interval_seconds will not be used.
I think we should notify the user by log entry that parameters passed are ignored.
Or we can do the other way around, if user passes waiter_countdown or waiter_check_interval_seconds we can force wait_for_completion to True even if user override this (We have similar case is in BaseOperator with wait_for_downstream and depends_on_past as far as I remember)

[syedahsn]

That's not entirely true. In both of the EMR Serverless operators, one call to the waiter function is made regardless of whether or not wait_for_completion is True. In this case, I think its better to leave it as is but I'm open to hearing other options.

[eladkal]

There is diffeent behavior for created and for started states.
Started state depends also on wait_for_completion

I'm not sure if functionality change is needed. That is your call
But I do think that clarification is required (doc wise?)
Given the code I shared earlier the parameters I passed will be consider for Created but not for Started.
I find this confusing. Do I miss something?

[syedahsn]

No you're correct in that when EmrServerlessCreateApplicationOperator is used, waiter can be called potentially twice, but definitely once. The first time, it will wait for the application to be in CREATED state, and it will use waiter_countdown and waiter_check_interval_seconds. If wait_for_completion is True, it will call waiter again and wait for the application to be in STARTED state.
For the EmrServerlessStartJobOperator, the waiter could potentially be called twice as well, but may not be called at all depending on whether the application is in a Started state or not. I added a change in the doc string to reflect this. Let me know if it is unclear. Thanks for the review!