Make RedshiftDeleteClusterOperator busy-retry window configurable

docs/spelling_wordlist.txt

@@ -447,6 +447,7 @@ deferrable
 deidentify
 DeidentifyTemplate
 del
+deletable
 delim
 deltalake
 denylist

providers/amazon/src/airflow/providers/amazon/aws/operators/redshift_cluster.py

@@ -27,6 +27,7 @@
 from airflow.providers.amazon.aws.hooks.redshift_cluster import RedshiftHook
 from airflow.providers.amazon.aws.operators.base_aws import AwsBaseOperator
 from airflow.providers.amazon.aws.triggers.redshift_cluster import (
+    RedshiftClusterSettledTrigger,
     RedshiftCreateClusterSnapshotTrigger,
     RedshiftCreateClusterTrigger,
     RedshiftDeleteClusterTrigger,
@@ -837,6 +838,14 @@ class RedshiftDeleteClusterOperator(AwsBaseOperator[RedshiftHook]):
     :param poll_interval: Time (in seconds) to wait between two consecutive calls to check cluster state
     :param deferrable: Run operator in the deferrable mode.
     :param max_attempts: (Deferrable mode only) The maximum number of attempts to be made
+    :param busy_retry_attempts: (Sync mode only) Number of times to retry the delete when the cluster is
+        busy and the request cannot yet be accepted (an ``InvalidClusterStateFault`` is raised), e.g.
+        while a pause/resume is in progress. The default of 60 (combined with ``busy_retry_interval``)
+        gives a ~15 minute window, long enough to outlast a cluster pause. This retry is synchronous and
+        blocks the worker; in deferrable mode the equivalent wait is performed asynchronously by the
+        triggerer instead (see ``deferrable``).
+    :param busy_retry_interval: (Sync mode only) Time (in seconds) to wait between busy retries when the
+        cluster is in a transient state (``InvalidClusterStateFault``). See ``busy_retry_attempts``.
     """
 
     template_fields: Sequence[str] = aws_template_fields(
@@ -856,6 +865,8 @@ def __init__(
         poll_interval: int = 30,
         deferrable: bool = conf.getboolean("operators", "default_deferrable", fallback=False),
         max_attempts: int = 30,
+        busy_retry_attempts: int = 60,
+        busy_retry_interval: int = 15,
         **kwargs,
     ):
         super().__init__(**kwargs)
@@ -864,16 +875,26 @@ def __init__(
         self.final_cluster_snapshot_identifier = final_cluster_snapshot_identifier
         self.wait_for_completion = wait_for_completion
         self.poll_interval = poll_interval
-        # These parameters are added to keep trying if there is a running operation in the cluster
-        # If there is a running operation in the cluster while trying to delete it, a InvalidClusterStateFault
-        # is thrown. In such case, retrying
-        self._attempts = 10
-        self._attempt_interval = 15
+        # These parameters are used to keep retrying if there is a running operation in the cluster.
+        # If there is a running operation in the cluster while trying to delete it (e.g. a pause/resume
+        # is in progress), an InvalidClusterStateFault is thrown. In such cases we retry until the
+        # cluster settles into a deletable state. The defaults give a ~15 minute window (60 * 15s),
+        # long enough to outlast a cluster pause.
+        self.busy_retry_attempts = busy_retry_attempts
+        self.busy_retry_interval = busy_retry_interval
         self.deferrable = deferrable
         self.max_attempts = max_attempts
 
     def execute(self, context: Context):
-        while self._attempts:
+        if self.deferrable:
+            # In deferrable mode we must not block the worker with the synchronous busy-retry loop.
+            # Attempt the delete once; if the cluster is mid-transition (InvalidClusterStateFault),
+            # hand off to the triggerer to wait for it to settle and then re-issue the delete.
+            self._delete_or_defer_until_settled()
+            return
+
+        attempts_remaining = self.busy_retry_attempts
+        while attempts_remaining:
             try:
                 self.hook.delete_cluster(
                     cluster_identifier=self.cluster_identifier,
@@ -882,51 +903,98 @@ def execute(self, context: Context):
                 )
                 break
             except self.hook.conn.exceptions.InvalidClusterStateFault:
-                self._attempts -= 1
+                attempts_remaining -= 1
 
-                if self._attempts:
+                if attempts_remaining:
                     current_state = self.hook.conn.describe_clusters(
                         ClusterIdentifier=self.cluster_identifier
                     )["Clusters"][0]["ClusterStatus"]
                     self.log.error(
                         "Cluster in %s state, unable to delete. %d attempts remaining.",
                         current_state,
-                        self._attempts,
+                        attempts_remaining,
                     )
-                    time.sleep(self._attempt_interval)
+                    time.sleep(self.busy_retry_interval)
                 else:
                     raise
 
-        if self.deferrable:
-            cluster_state = self.hook.cluster_status(cluster_identifier=self.cluster_identifier)
-            if cluster_state == "cluster_not_found":
-                self.log.info("Cluster deleted successfully")
-            elif cluster_state in ("creating", "modifying"):
-                raise AirflowException(
-                    f"Unable to delete cluster since cluster is currently in status: {cluster_state}"
-                )
-            else:
-                self.defer(
-                    timeout=timedelta(seconds=self.max_attempts * self.poll_interval + 60),
-                    trigger=RedshiftDeleteClusterTrigger(
-                        cluster_identifier=self.cluster_identifier,
-                        waiter_delay=self.poll_interval,
-                        waiter_max_attempts=self.max_attempts,
-                        aws_conn_id=self.aws_conn_id,
-                        region_name=self.region_name,
-                        verify=self.verify,
-                        botocore_config=self.botocore_config,
-                    ),
-                    method_name="execute_complete",
-                )
-
-        elif self.wait_for_completion:
+        if self.wait_for_completion:
             waiter = self.hook.conn.get_waiter("cluster_deleted")
             waiter.wait(
                 ClusterIdentifier=self.cluster_identifier,
                 WaiterConfig={"Delay": self.poll_interval, "MaxAttempts": self.max_attempts},
             )
 
+    def _delete_or_defer_until_settled(self) -> None:
+        """
+        Issue the delete once (deferrable mode); defer to wait out a busy cluster if needed.
+
+        If the delete is accepted, defer to :class:`RedshiftDeleteClusterTrigger` to wait for the
+        deletion to finish. If the cluster is mid-transition (``InvalidClusterStateFault``), defer to
+        :class:`RedshiftClusterSettledTrigger`, which fires once the cluster leaves every transitional
+        lifecycle; the ``_retry_delete_when_settled`` callback then re-issues the delete.
+        """
+        try:
+            self.hook.delete_cluster(
+                cluster_identifier=self.cluster_identifier,
+                skip_final_cluster_snapshot=self.skip_final_cluster_snapshot,
+                final_cluster_snapshot_identifier=self.final_cluster_snapshot_identifier,
+            )
+        except self.hook.conn.exceptions.InvalidClusterStateFault:
+            self.log.info(
+                "Cluster %s is busy; deferring until it settles into a deletable state.",
+                self.cluster_identifier,
+            )
+            self.defer(
+                timeout=timedelta(seconds=self.busy_retry_attempts * self.busy_retry_interval + 60),
+                trigger=RedshiftClusterSettledTrigger(
+                    cluster_identifier=self.cluster_identifier,
+                    poke_interval=self.busy_retry_interval,
+                    max_attempts=self.busy_retry_attempts,
+                    aws_conn_id=self.aws_conn_id,
+                    region_name=self.region_name,
+                    verify=self.verify,
+                    botocore_config=self.botocore_config,
+                ),
+                method_name="_retry_delete_when_settled",
+            )
+            return
+
+        self._defer_until_deleted()
+
+    def _defer_until_deleted(self) -> None:
+        """Defer to the delete-completion waiter, short-circuiting if the cluster is already gone."""
+        cluster_state = self.hook.cluster_status(cluster_identifier=self.cluster_identifier)
+        if cluster_state == "cluster_not_found":
+            self.log.info("Cluster deleted successfully")
+            return
+        self.defer(
+            timeout=timedelta(seconds=self.max_attempts * self.poll_interval + 60),
+            trigger=RedshiftDeleteClusterTrigger(
+                cluster_identifier=self.cluster_identifier,
+                waiter_delay=self.poll_interval,
+                waiter_max_attempts=self.max_attempts,
+                aws_conn_id=self.aws_conn_id,
+                region_name=self.region_name,
+                verify=self.verify,
+                botocore_config=self.botocore_config,
+            ),
+            method_name="execute_complete",
+        )
+
+    def _retry_delete_when_settled(self, context: Context, event: dict[str, Any] | None = None) -> None:
+        """
+        Re-issue the delete once the cluster has settled, then defer until deletion completes.
+
+        Callback for :class:`RedshiftClusterSettledTrigger`. If the delete is still rejected because of a
+        race (the cluster re-entered a transitional state), defer to the settle-wait trigger again.
+        """
+        validated_event = validate_execute_complete_event(event)
+        if validated_event["status"] != "success":
+            raise AirflowException(f"Error waiting for cluster to become deletable: {validated_event}")
+
+        self._delete_or_defer_until_settled()
+
     def execute_complete(self, context: Context, event: dict[str, Any] | None = None) -> None:
         validated_event = validate_execute_complete_event(event)
 

providers/amazon/src/airflow/providers/amazon/aws/triggers/redshift_cluster.py

@@ -28,6 +28,28 @@
 if TYPE_CHECKING:
     from airflow.providers.amazon.aws.hooks.base_aws import AwsGenericHook
 
+# Cluster lifecycle states during which a ``delete_cluster`` call is rejected with an
+# ``InvalidClusterStateFault`` because an operation is already in flight. Once the cluster leaves all
+# of these states it has "settled" into a deletable lifecycle (e.g. ``available`` or ``paused``) and the
+# delete can be re-issued.
+REDSHIFT_TRANSITIONAL_CLUSTER_STATES = frozenset(
+    {
+        "creating",
+        "deleting",
+        "modifying",
+        "rebooting",
+        "renaming",
+        "resizing",
+        "resuming",
+        "pausing",
+        "rotating-keys",
+        "updating-hsm",
+        "cancelling-resize",
+        "prep-for-resize",
+        "final-snapshot",
+    }
+)
+
 
 class RedshiftCreateClusterTrigger(AwsBaseWaiterTrigger):
     """
@@ -343,3 +365,90 @@ async def run(self) -> AsyncIterator[TriggerEvent]:
                 await asyncio.sleep(self.poke_interval)
         except Exception as e:
             yield TriggerEvent({"status": "error", "message": str(e)})
+
+
+class RedshiftClusterSettledTrigger(BaseTrigger):
+    """
+    Poll a Redshift cluster until it settles into a non-transitional (deletable) lifecycle.
+
+    A ``delete_cluster`` call is rejected with ``InvalidClusterStateFault`` while an operation is in
+    flight (e.g. a pause or resize). Because a busy cluster can settle into *different* terminal states
+    depending on the in-flight operation (a ``pausing`` cluster becomes ``paused``; a ``resizing`` cluster
+    becomes ``available``), this trigger fires as soon as the status leaves every transitional state
+    rather than waiting for a single hardcoded ``target_status``. The operator then re-issues the delete.
+
+    :param aws_conn_id: Reference to AWS connection id for redshift
+    :param cluster_identifier: unique identifier of a cluster
+    :param poke_interval: polling period in seconds to check for the status
+    :param max_attempts: maximum number of polls before emitting an error event
+    :param region_name: The AWS region where the cluster is. Used to build the hook.
+    :param verify: Whether or not to verify SSL certificates. Used to build the hook.
+    :param botocore_config: Configuration dictionary for the botocore client. Used to build the hook.
+    """
+
+    def __init__(
+        self,
+        *,
+        aws_conn_id: str | None,
+        cluster_identifier: str,
+        poke_interval: float = 30,
+        max_attempts: int = 30,
+        region_name: str | None = None,
+        verify: bool | str | None = None,
+        botocore_config: dict | None = None,
+    ):
+        super().__init__()
+        self.aws_conn_id = aws_conn_id
+        self.cluster_identifier = cluster_identifier
+        self.poke_interval = poke_interval
+        self.max_attempts = max_attempts
+        self.region_name = region_name
+        self.verify = verify
+        self.botocore_config = botocore_config
+
+    def serialize(self) -> tuple[str, dict[str, Any]]:
+        """Serialize RedshiftClusterSettledTrigger arguments and classpath."""
+        return (
+            "airflow.providers.amazon.aws.triggers.redshift_cluster.RedshiftClusterSettledTrigger",
+            {
+                "aws_conn_id": self.aws_conn_id,
+                "cluster_identifier": self.cluster_identifier,
+                "poke_interval": self.poke_interval,
+                "max_attempts": self.max_attempts,
+                "region_name": self.region_name,
+                "verify": self.verify,
+                "botocore_config": self.botocore_config,
+            },
+        )
+
+    @cached_property
+    def hook(self) -> RedshiftHook:
+        return RedshiftHook(
+            aws_conn_id=self.aws_conn_id,
+            region_name=self.region_name,
+            verify=self.verify,
+            config=self.botocore_config,
+        )
+
+    async def run(self) -> AsyncIterator[TriggerEvent]:
+        """Run async until the cluster leaves every transitional state (or is already gone)."""
+        try:
+            for _ in range(self.max_attempts):
+                status = await self.hook.cluster_status_async(self.cluster_identifier)
+                if status is None or status.lower() not in REDSHIFT_TRANSITIONAL_CLUSTER_STATES:
+                    yield TriggerEvent(
+                        {"status": "success", "message": "Cluster settled", "cluster_state": status}
+                    )
+                    return
+                await asyncio.sleep(self.poke_interval)
+            yield TriggerEvent(
+                {
+                    "status": "error",
+                    "message": (
+                        f"Cluster {self.cluster_identifier} did not settle into a deletable state "
+                        f"within {self.max_attempts} attempts."
+                    ),
+                }
+            )
+        except Exception as e:
+            yield TriggerEvent({"status": "error", "message": str(e)})