vandonr-amz commented on code in PR #30463:
URL: https://github.com/apache/airflow/pull/30463#discussion_r1159155707
##########
airflow/providers/amazon/aws/operators/emr.py:
##########
@@ -809,10 +810,13 @@ class
EmrServerlessCreateApplicationOperator(BaseOperator):
Its value must be unique for each request.
:param config: Optional dictionary for arbitrary parameters to the boto
API create_application call.
:param aws_conn_id: AWS connection to use
- :param waiter_countdown: Total amount of time, in seconds, the operator
will wait for
+ :param waiter_countdown: (deprecated) Total amount of time, in seconds,
the operator will wait for
the application to start. Defaults to 25 minutes.
- :param waiter_check_interval_seconds: Number of seconds between polling
the state of the application.
- Defaults to 60 seconds.
+ :param waiter_check_interval_seconds: (deprecated) Number of seconds
between polling the state
Review Comment:
Not super fan of making user change their code for no reason except to
standardize names.
This is adding friction to upgrading where we could very well keep the
current name since there is no behavior change.
##########
airflow/providers/amazon/aws/operators/emr.py:
##########
@@ -992,21 +1036,40 @@ def execute(self, context: Context) -> dict:
self.log.info("EMR serverless job started: %s", response["jobRunId"])
if self.wait_for_completion:
- # This should be replaced with a boto waiter when available.
- waiter(
- get_state_callable=self.hook.conn.get_job_run,
- get_state_args={
- "applicationId": self.application_id,
- "jobRunId": response["jobRunId"],
- },
- parse_response=["jobRun", "state"],
- desired_state=EmrServerlessHook.JOB_SUCCESS_STATES,
- failure_states=EmrServerlessHook.JOB_FAILURE_STATES,
- object_type="job",
- action="run",
- countdown=self.waiter_countdown,
- check_interval_seconds=self.waiter_check_interval_seconds,
- )
+ try:
+ self.hook.get_waiter("serverless_job_running").wait(
+ applicationId=self.application_id,
+ jobRunId=response["jobRunId"],
+ WaiterConfig=prune_dict(
+ {
+ "Delay": self.waiter_delay,
+ "MaxAttempts": self.waiter_max_attempts,
+ }
+ ),
+ )
+ self.log.info("EMR serverless job is running: %s",
response["jobRunId"])
+ self.hook.get_waiter("serverless_job_completed").wait(
+ applicationId=self.application_id,
+ jobRunId=response["jobRunId"],
+ WaiterConfig=prune_dict(
+ {
+ "Delay": self.waiter_delay,
+ "MaxAttempts": self.waiter_max_attempts,
+ }
+ ),
+ )
Review Comment:
does this really need to be 2 waiters just to be able to log a line when the
job is running ? Doesn't the waiter already log what state it's seeing ?
##########
airflow/providers/amazon/aws/operators/emr.py:
##########
@@ -823,18 +827,41 @@ def __init__(
config: dict | None = None,
wait_for_completion: bool = True,
aws_conn_id: str = "aws_default",
- waiter_countdown: int = 25 * 60,
- waiter_check_interval_seconds: int = 60,
+ waiter_max_attempts: int | None | ArgNotSet = NOTSET,
+ waiter_delay: int | None | ArgNotSet = NOTSET,
Review Comment:
why allow None here ? It's either set to an integer value or not set, isn't
it ?
(same question for other constructors below)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
