Resolving EMR notebook deprecated warnings

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

@@ -393,34 +393,38 @@ def __init__(
         # TODO: waiter_max_attempts and waiter_delay should default to None when the other two are deprecated.
         waiter_max_attempts: int | None | ArgNotSet = NOTSET,
         waiter_delay: int | None | ArgNotSet = NOTSET,
-        waiter_countdown: int = 25 * 60,
-        waiter_check_interval_seconds: int = 60,
+        waiter_countdown: int | None = None,
+        waiter_check_interval_seconds: int | None = None,
         **kwargs: Any,
     ):
-        if waiter_max_attempts is NOTSET:
+        if waiter_check_interval_seconds:
             warnings.warn(
-                "The parameter waiter_countdown has been deprecated to standardize "
-                "naming conventions.  Please use waiter_max_attempts instead.  In the "
+                "The parameter `waiter_check_interval_seconds` has been deprecated to "
+                "standardize naming conventions.  Please `use waiter_delay instead`.  In the "
                 "future this will default to None and defer to the waiter's default value.",
                 AirflowProviderDeprecationWarning,
                 stacklevel=2,
             )
-            waiter_max_attempts = waiter_countdown // waiter_check_interval_seconds
-        if waiter_delay is NOTSET:
+        else:
+            waiter_check_interval_seconds = 60
+        if waiter_countdown:
             warnings.warn(
-                "The parameter waiter_check_interval_seconds has been deprecated to "
-                "standardize naming conventions.  Please use waiter_delay instead.  In the "
+                "The parameter waiter_countdown has been deprecated to standardize "
+                "naming conventions.  Please use waiter_max_attempts instead.  In the "
                 "future this will default to None and defer to the waiter's default value.",
                 AirflowProviderDeprecationWarning,
                 stacklevel=2,
             )
-            waiter_delay = waiter_check_interval_seconds
+            # waiter_countdown defaults to never timing out, which is not supported
+            # by boto waiters, so we will set it here to "a very long time" for now.
+            waiter_max_attempts = (waiter_countdown or 999) // waiter_check_interval_seconds
+
         super().__init__(**kwargs)
         self.notebook_execution_id = notebook_execution_id
         self.wait_for_completion = wait_for_completion
         self.aws_conn_id = aws_conn_id
-        self.waiter_max_attempts = waiter_max_attempts
-        self.waiter_delay = waiter_delay
+        self.waiter_max_attempts = waiter_max_attempts or 25
+        self.waiter_delay = waiter_delay or waiter_check_interval_seconds or 60
 
     def execute(self, context: Context) -> None:
         emr_hook = EmrHook(aws_conn_id=self.aws_conn_id)