From bf712ff6fc3a097f008fe9dd65b854c7b6d51b4f Mon Sep 17 00:00:00 2001 From: Sean Ghaeli Date: Tue, 7 Jul 2026 21:26:27 +0000 Subject: [PATCH 1/7] Make RedshiftDeleteClusterOperator delete reliably during cluster transitions A delete issued while the cluster is mid-transition (pausing/resuming/resizing) raises InvalidClusterStateFault. The retry budget was hardcoded to 10 * 15s = 2.5 min, which expires long before a real transition settles (~minutes), so the task fails and the cluster leaks. Non-deferrable mode: raise the synchronous busy-retry budget to 60 * 15s = ~15 min so it outlasts a transition; still fail-loud once exhausted. Deferrable mode: previously the synchronous loop ran before the defer, blocking the worker. Now attempt the delete once; on InvalidClusterStateFault defer to a new RedshiftClusterSettledTrigger (an AwsBaseWaiterTrigger backed by a custom cluster_deletable waiter that treats transitional states as retry and fires once the cluster is deletable), then a callback re-issues the delete and defers to the existing RedshiftDeleteClusterTrigger; re-defers on a race. Bounded by the existing poll_interval/max_attempts. This mirrors the EKS deferrable re-defer pattern and the AwsBaseWaiterTrigger convention used by the other Redshift triggers. No worker is blocked in deferrable mode. Generated-by: Claude Code (Opus) --- .../amazon/aws/operators/redshift_cluster.py | 115 ++++++++--- .../amazon/aws/triggers/redshift_cluster.py | 56 ++++++ .../amazon/aws/waiters/redshift.json | 55 ++++++ .../aws/operators/test_redshift_cluster.py | 178 +++++++++++++++++- .../aws/triggers/test_redshift_cluster.py | 36 +++- 5 files changed, 394 insertions(+), 46 deletions(-) diff --git a/providers/amazon/src/airflow/providers/amazon/aws/operators/redshift_cluster.py b/providers/amazon/src/airflow/providers/amazon/aws/operators/redshift_cluster.py index ea0d1272db8c6..1c2c2f6ffec93 100644 --- a/providers/amazon/src/airflow/providers/amazon/aws/operators/redshift_cluster.py +++ b/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, @@ -836,7 +837,9 @@ class RedshiftDeleteClusterOperator(AwsBaseOperator[RedshiftHook]): https://boto3.amazonaws.com/v1/documentation/api/latest/reference/core/session.html :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 max_attempts: The maximum number of attempts to be made. In deferrable mode this bounds the + async wait for a busy cluster to settle before the delete is re-issued; combined with + ``poll_interval`` the default gives a ~15 minute window, long enough to outlast a pause/resize. """ template_fields: Sequence[str] = aws_template_fields( @@ -864,15 +867,22 @@ 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 + # Keep retrying while another operation is running on the cluster: a delete issued mid-transition + # (e.g. a pause/resize in progress) raises InvalidClusterStateFault. Retry until the cluster + # settles into a deletable state. 60 * 15s = ~15 min, long enough to outlast a cluster pause. + self._attempts = 60 self._attempt_interval = 15 self.deferrable = deferrable self.max_attempts = max_attempts def execute(self, context: Context): + 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 + while self._attempts: try: self.hook.delete_cluster( @@ -897,36 +907,83 @@ def execute(self, context: Context): 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.max_attempts * self.poll_interval + 60), + trigger=RedshiftClusterSettledTrigger( + 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="_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) diff --git a/providers/amazon/src/airflow/providers/amazon/aws/triggers/redshift_cluster.py b/providers/amazon/src/airflow/providers/amazon/aws/triggers/redshift_cluster.py index ebd32a2b42388..1f12e83192835 100644 --- a/providers/amazon/src/airflow/providers/amazon/aws/triggers/redshift_cluster.py +++ b/providers/amazon/src/airflow/providers/amazon/aws/triggers/redshift_cluster.py @@ -343,3 +343,59 @@ 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(AwsBaseWaiterTrigger): + """ + Wait until a Redshift cluster 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 uses the custom ``cluster_deletable`` waiter, which has one + ``success`` acceptor per deletable ``ClusterStatus`` value. Any transitional state (``pausing``, + ``resizing``, ``resuming``, ``modifying``, ...) implicitly retries. A missing cluster + (``ClusterNotFound``) is treated as success since there is nothing left to wait on. The operator then + re-issues the delete. + + :param cluster_identifier: unique identifier of a cluster + :param waiter_delay: The amount of time in seconds to wait between attempts. + :param waiter_max_attempts: The maximum number of attempts to be made. + :param aws_conn_id: The Airflow connection used for AWS credentials. + :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, + *, + cluster_identifier: str, + aws_conn_id: str | None = "aws_default", + region_name: str | None = None, + waiter_delay: int = 30, + waiter_max_attempts: int = 30, + **kwargs, + ): + super().__init__( + serialized_fields={"cluster_identifier": cluster_identifier}, + waiter_name="cluster_deletable", + waiter_args={"ClusterIdentifier": cluster_identifier}, + failure_message="Error while waiting for the redshift cluster to become deletable", + status_message="Waiting for redshift cluster to settle into a deletable state", + status_queries=["Clusters[].ClusterStatus"], + return_value=None, + waiter_delay=waiter_delay, + waiter_max_attempts=waiter_max_attempts, + aws_conn_id=aws_conn_id, + region_name=region_name, + **kwargs, + ) + + def hook(self) -> AwsGenericHook: + return RedshiftHook( + aws_conn_id=self.aws_conn_id, + region_name=self.region_name, + verify=self.verify, + config=self.botocore_config, + ) diff --git a/providers/amazon/src/airflow/providers/amazon/aws/waiters/redshift.json b/providers/amazon/src/airflow/providers/amazon/aws/waiters/redshift.json index 8165eb3fc439a..389da67068e90 100644 --- a/providers/amazon/src/airflow/providers/amazon/aws/waiters/redshift.json +++ b/providers/amazon/src/airflow/providers/amazon/aws/waiters/redshift.json @@ -50,6 +50,61 @@ "state": "failure" } ] + }, + "cluster_deletable": { + "operation": "DescribeClusters", + "delay": 30, + "maxAttempts": 60, + "acceptors": [ + { + "matcher": "pathAll", + "argument": "Clusters[].ClusterStatus", + "expected": "available", + "state": "success" + }, + { + "matcher": "pathAll", + "argument": "Clusters[].ClusterStatus", + "expected": "paused", + "state": "success" + }, + { + "matcher": "pathAll", + "argument": "Clusters[].ClusterStatus", + "expected": "incompatible-hsm", + "state": "success" + }, + { + "matcher": "pathAll", + "argument": "Clusters[].ClusterStatus", + "expected": "incompatible-restore", + "state": "success" + }, + { + "matcher": "pathAll", + "argument": "Clusters[].ClusterStatus", + "expected": "incompatible-network", + "state": "success" + }, + { + "matcher": "pathAll", + "argument": "Clusters[].ClusterStatus", + "expected": "incompatible-parameters", + "state": "success" + }, + { + "matcher": "pathAll", + "argument": "Clusters[].ClusterStatus", + "expected": "hardware-failure", + "state": "success" + }, + { + "matcher": "error", + "argument": "Clusters[].ClusterStatus", + "expected": "ClusterNotFound", + "state": "success" + } + ] } } } diff --git a/providers/amazon/tests/unit/amazon/aws/operators/test_redshift_cluster.py b/providers/amazon/tests/unit/amazon/aws/operators/test_redshift_cluster.py index db6466125a035..c329b9431dd97 100644 --- a/providers/amazon/tests/unit/amazon/aws/operators/test_redshift_cluster.py +++ b/providers/amazon/tests/unit/amazon/aws/operators/test_redshift_cluster.py @@ -34,6 +34,7 @@ RedshiftResumeClusterOperator, ) from airflow.providers.amazon.aws.triggers.redshift_cluster import ( + RedshiftClusterSettledTrigger, RedshiftCreateClusterSnapshotTrigger, RedshiftDeleteClusterTrigger, RedshiftPauseClusterTrigger, @@ -785,14 +786,65 @@ def test_delete_cluster_multiple_attempts_fail(self, _, mock_conn, mock_delete_c with pytest.raises(returned_exception): redshift_operator.execute(None) - assert mock_delete_cluster.call_count == 10 + assert mock_delete_cluster.call_count == 60 + + def test_busy_retry_defaults(self): + redshift_operator = RedshiftDeleteClusterOperator( + task_id="task_test", + cluster_identifier="test_cluster", + aws_conn_id="aws_conn_test", + ) + # 60 * 15s = ~15 min window, long enough to outlast a cluster pause/resize. + assert redshift_operator._attempts == 60 + assert redshift_operator._attempt_interval == 15 + + @mock.patch.object(RedshiftHook, "delete_cluster") + @mock.patch.object(RedshiftHook, "conn") + @mock.patch("time.sleep", return_value=None) + def test_delete_cluster_exhausts_busy_retries_then_raises(self, _, mock_conn, mock_delete_cluster): + exception = boto3.client("redshift").exceptions.InvalidClusterStateFault({}, "test") + returned_exception = type(exception) + mock_conn.exceptions.InvalidClusterStateFault = returned_exception + mock_delete_cluster.side_effect = exception + + redshift_operator = RedshiftDeleteClusterOperator( + task_id="task_test", + cluster_identifier="test_cluster", + aws_conn_id="aws_conn_test", + wait_for_completion=False, + ) + redshift_operator._attempts = 3 + with pytest.raises(returned_exception): + redshift_operator.execute(None) + + assert mock_delete_cluster.call_count == 3 + + @mock.patch.object(RedshiftHook, "delete_cluster") + @mock.patch.object(RedshiftHook, "conn") + @mock.patch("time.sleep", return_value=None) + def test_delete_cluster_succeeds_on_second_attempt(self, _, mock_conn, mock_delete_cluster): + exception = boto3.client("redshift").exceptions.InvalidClusterStateFault({}, "test") + returned_exception = type(exception) + mock_conn.exceptions.InvalidClusterStateFault = returned_exception + mock_delete_cluster.side_effect = [exception, True] + + redshift_operator = RedshiftDeleteClusterOperator( + task_id="task_test", + cluster_identifier="test_cluster", + aws_conn_id="aws_conn_test", + wait_for_completion=False, + ) + redshift_operator._attempt_interval = 0 + redshift_operator.execute(None) + + assert mock_delete_cluster.call_count == 2 @mock.patch("airflow.providers.amazon.aws.hooks.redshift_cluster.RedshiftHook.cluster_status") @mock.patch("airflow.providers.amazon.aws.hooks.redshift_cluster.RedshiftHook.delete_cluster") def test_delete_cluster_deferrable_mode(self, mock_delete_cluster, mock_cluster_status): - """Test delete cluster operator with defer when deferrable param is true""" + """When the delete is accepted, deferrable mode waits for deletion to complete.""" mock_delete_cluster.return_value = True - mock_cluster_status.return_value = "available" + mock_cluster_status.return_value = "deleting" delete_cluster = RedshiftDeleteClusterOperator( task_id="task_test", cluster_identifier="test_cluster", @@ -806,16 +858,82 @@ def test_delete_cluster_deferrable_mode(self, mock_delete_cluster, mock_cluster_ assert isinstance(exc.value.trigger, RedshiftDeleteClusterTrigger), ( "Trigger is not a RedshiftDeleteClusterTrigger" ) + # Delete is attempted exactly once (no synchronous busy-retry loop in deferrable mode). + mock_delete_cluster.assert_called_once() - @mock.patch("airflow.providers.amazon.aws.operators.redshift_cluster.RedshiftDeleteClusterOperator.defer") @mock.patch("airflow.providers.amazon.aws.hooks.redshift_cluster.RedshiftHook.cluster_status") @mock.patch("airflow.providers.amazon.aws.hooks.redshift_cluster.RedshiftHook.delete_cluster") - def test_delete_cluster_deferrable_mode_in_paused_state( - self, mock_delete_cluster, mock_cluster_status, mock_defer + def test_delete_cluster_deferrable_mode_already_gone(self, mock_delete_cluster, mock_cluster_status): + """When the cluster is already gone after the delete, deferrable mode completes without deferring.""" + mock_delete_cluster.return_value = True + mock_cluster_status.return_value = "cluster_not_found" + delete_cluster = RedshiftDeleteClusterOperator( + task_id="task_test", + cluster_identifier="test_cluster", + deferrable=True, + wait_for_completion=False, + ) + + # No TaskDeferred is raised; the operator returns normally. + delete_cluster.execute(context=None) + mock_delete_cluster.assert_called_once() + + @mock.patch.object(RedshiftHook, "conn") + @mock.patch("airflow.providers.amazon.aws.hooks.redshift_cluster.RedshiftHook.delete_cluster") + def test_delete_cluster_deferrable_mode_busy_defers_to_settle_trigger( + self, mock_delete_cluster, mock_conn ): - """Test delete cluster operator with defer when deferrable param is true""" + """A busy cluster (InvalidClusterStateFault) defers to the settle-wait trigger, not a sync loop.""" + exception = boto3.client("redshift").exceptions.InvalidClusterStateFault({}, "test") + mock_conn.exceptions.InvalidClusterStateFault = type(exception) + mock_delete_cluster.side_effect = exception + + delete_cluster = RedshiftDeleteClusterOperator( + task_id="task_test", + cluster_identifier="test_cluster", + deferrable=True, + wait_for_completion=False, + ) + + with pytest.raises(TaskDeferred) as exc: + delete_cluster.execute(context=None) + + assert isinstance(exc.value.trigger, RedshiftClusterSettledTrigger), ( + "Trigger is not a RedshiftClusterSettledTrigger" + ) + assert exc.value.method_name == "_retry_delete_when_settled" + # Delete attempted once; the synchronous busy-retry loop never runs in deferrable mode. + mock_delete_cluster.assert_called_once() + + @mock.patch("airflow.providers.amazon.aws.hooks.redshift_cluster.RedshiftHook.cluster_status") + @mock.patch("airflow.providers.amazon.aws.hooks.redshift_cluster.RedshiftHook.delete_cluster") + def test_retry_delete_when_settled_reissues_delete(self, mock_delete_cluster, mock_cluster_status): + """The settle-wait callback re-issues the delete and defers to the delete-complete trigger.""" mock_delete_cluster.return_value = True - mock_cluster_status.return_value = "creating" + mock_cluster_status.return_value = "deleting" + delete_cluster = RedshiftDeleteClusterOperator( + task_id="task_test", + cluster_identifier="test_cluster", + deferrable=True, + wait_for_completion=False, + ) + + with pytest.raises(TaskDeferred) as exc: + delete_cluster._retry_delete_when_settled( + context=None, event={"status": "success", "message": "Cluster settled"} + ) + + assert isinstance(exc.value.trigger, RedshiftDeleteClusterTrigger) + mock_delete_cluster.assert_called_once() + + @mock.patch.object(RedshiftHook, "conn") + @mock.patch("airflow.providers.amazon.aws.hooks.redshift_cluster.RedshiftHook.delete_cluster") + def test_retry_delete_when_settled_redefers_on_race(self, mock_delete_cluster, mock_conn): + """If the cluster re-enters a transitional state (race), the callback re-defers to settle-wait.""" + exception = boto3.client("redshift").exceptions.InvalidClusterStateFault({}, "test") + mock_conn.exceptions.InvalidClusterStateFault = type(exception) + mock_delete_cluster.side_effect = exception + delete_cluster = RedshiftDeleteClusterOperator( task_id="task_test", cluster_identifier="test_cluster", @@ -823,10 +941,50 @@ def test_delete_cluster_deferrable_mode_in_paused_state( wait_for_completion=False, ) + with pytest.raises(TaskDeferred) as exc: + delete_cluster._retry_delete_when_settled( + context=None, event={"status": "success", "message": "Cluster settled"} + ) + + assert isinstance(exc.value.trigger, RedshiftClusterSettledTrigger) + assert exc.value.method_name == "_retry_delete_when_settled" + + def test_retry_delete_when_settled_error_event_raises(self): + """A non-success event from the settle-wait trigger raises AirflowException.""" + delete_cluster = RedshiftDeleteClusterOperator( + task_id="task_test", + cluster_identifier="test_cluster", + deferrable=True, + wait_for_completion=False, + ) with pytest.raises(AirflowException): - delete_cluster.execute(context=None) + delete_cluster._retry_delete_when_settled( + context=None, event={"status": "error", "message": "timed out"} + ) - assert not mock_defer.called + @mock.patch.object(RedshiftHook, "delete_cluster") + @mock.patch.object(RedshiftHook, "conn") + @mock.patch("time.sleep", return_value=None) + def test_delete_cluster_sync_mode_still_uses_busy_retry_loop( + self, mock_sleep, mock_conn, mock_delete_cluster + ): + """Sync mode (deferrable=False) must keep the synchronous busy-retry loop unchanged.""" + exception = boto3.client("redshift").exceptions.InvalidClusterStateFault({}, "test") + mock_conn.exceptions.InvalidClusterStateFault = type(exception) + mock_delete_cluster.side_effect = [exception, exception, True] + + redshift_operator = RedshiftDeleteClusterOperator( + task_id="task_test", + cluster_identifier="test_cluster", + aws_conn_id="aws_conn_test", + deferrable=False, + wait_for_completion=False, + ) + redshift_operator.execute(None) + + # Three synchronous attempts, and time.sleep was used between the retries. + assert mock_delete_cluster.call_count == 3 + assert mock_sleep.called def test_delete_cluster_execute_complete_success(self): """Asserts that logging occurs as expected""" diff --git a/providers/amazon/tests/unit/amazon/aws/triggers/test_redshift_cluster.py b/providers/amazon/tests/unit/amazon/aws/triggers/test_redshift_cluster.py index a55494ce8ab7c..42f0b5f1f805d 100644 --- a/providers/amazon/tests/unit/amazon/aws/triggers/test_redshift_cluster.py +++ b/providers/amazon/tests/unit/amazon/aws/triggers/test_redshift_cluster.py @@ -23,6 +23,7 @@ import pytest from airflow.providers.amazon.aws.triggers.redshift_cluster import ( + RedshiftClusterSettledTrigger, RedshiftClusterTrigger, RedshiftCreateClusterSnapshotTrigger, RedshiftCreateClusterTrigger, @@ -152,43 +153,64 @@ async def test_redshift_cluster_sensor_trigger_exception(self, mock_cluster_stat RedshiftCreateClusterTrigger, 15, 999999, + "cluster_available", id="RedshiftCreateClusterTrigger", ), pytest.param( RedshiftPauseClusterTrigger, 15, 999999, + "cluster_paused", id="RedshiftPauseClusterTrigger", ), pytest.param( RedshiftCreateClusterSnapshotTrigger, 15, 999999, + "snapshot_available", id="RedshiftCreateClusterSnapshotTrigger", ), pytest.param( RedshiftResumeClusterTrigger, 15, 999999, + "cluster_resumed", id="RedshiftResumeClusterTrigger", ), pytest.param( RedshiftDeleteClusterTrigger, 30, 30, + "cluster_deleted", id="RedshiftDeleteClusterTrigger", ), + pytest.param( + RedshiftClusterSettledTrigger, + 30, + 30, + "cluster_deletable", + id="RedshiftClusterSettledTrigger", + ), ] class TestRedshiftWaiterTriggers: - """Tests for the five Redshift triggers that inherit from ``AwsBaseWaiterTrigger``.""" + """Tests for the Redshift triggers that inherit from ``AwsBaseWaiterTrigger``.""" + + @pytest.mark.parametrize( + ("trigger_cls", "default_delay", "default_max_attempts", "waiter_name"), + WAITER_TRIGGER_PARAMS, + ) + def test_waiter_name(self, trigger_cls, default_delay, default_max_attempts, waiter_name): + trigger = trigger_cls(cluster_identifier="test_cluster") + assert trigger.waiter_name == waiter_name + assert trigger.waiter_args == {"ClusterIdentifier": "test_cluster"} @pytest.mark.parametrize( - ("trigger_cls", "default_delay", "default_max_attempts"), + ("trigger_cls", "default_delay", "default_max_attempts", "waiter_name"), WAITER_TRIGGER_PARAMS, ) - def test_serialization(self, trigger_cls, default_delay, default_max_attempts): + def test_serialization(self, trigger_cls, default_delay, default_max_attempts, waiter_name): trigger = trigger_cls( cluster_identifier="test_cluster", aws_conn_id="aws_default", @@ -206,11 +228,11 @@ def test_serialization(self, trigger_cls, default_delay, default_max_attempts): } @pytest.mark.parametrize( - ("trigger_cls", "default_delay", "default_max_attempts"), + ("trigger_cls", "default_delay", "default_max_attempts", "waiter_name"), WAITER_TRIGGER_PARAMS, ) def test_serialization_with_verify_and_botocore_config( - self, trigger_cls, default_delay, default_max_attempts + self, trigger_cls, default_delay, default_max_attempts, waiter_name ): trigger = trigger_cls( cluster_identifier="test_cluster", @@ -225,12 +247,12 @@ def test_serialization_with_verify_and_botocore_config( assert "region_name" not in kwargs @pytest.mark.parametrize( - ("trigger_cls", "default_delay", "default_max_attempts"), + ("trigger_cls", "default_delay", "default_max_attempts", "waiter_name"), WAITER_TRIGGER_PARAMS, ) @mock.patch("airflow.providers.amazon.aws.triggers.redshift_cluster.RedshiftHook") def test_hook_propagates_verify_and_botocore_config( - self, mock_hook_cls, trigger_cls, default_delay, default_max_attempts + self, mock_hook_cls, trigger_cls, default_delay, default_max_attempts, waiter_name ): trigger = trigger_cls( cluster_identifier="test_cluster", From 230a2325889bd3e975c8225e6faec70bf7c0519f Mon Sep 17 00:00:00 2001 From: Sean Ghaeli Date: Tue, 7 Jul 2026 21:38:03 +0000 Subject: [PATCH 2/7] Trim verbose docstrings in redshift delete operator/trigger Comment-only: tighten the RedshiftClusterSettledTrigger class docstring and the _delete_or_defer_until_settled docstring, keeping the non-obvious rationale (multi-terminal-state -> custom waiter, which trigger handles which case). Generated-by: Claude Code (Opus) --- .../providers/amazon/aws/operators/redshift_cluster.py | 9 ++++----- .../providers/amazon/aws/triggers/redshift_cluster.py | 5 +---- 2 files changed, 5 insertions(+), 9 deletions(-) diff --git a/providers/amazon/src/airflow/providers/amazon/aws/operators/redshift_cluster.py b/providers/amazon/src/airflow/providers/amazon/aws/operators/redshift_cluster.py index 1c2c2f6ffec93..9d993deca72f1 100644 --- a/providers/amazon/src/airflow/providers/amazon/aws/operators/redshift_cluster.py +++ b/providers/amazon/src/airflow/providers/amazon/aws/operators/redshift_cluster.py @@ -916,12 +916,11 @@ def execute(self, context: Context): def _delete_or_defer_until_settled(self) -> None: """ - Issue the delete once (deferrable mode); defer to wait out a busy cluster if needed. + Issue the delete once (deferrable mode), then defer. - 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. + If accepted, defer to :class:`RedshiftDeleteClusterTrigger` to await deletion. If the cluster is + busy (``InvalidClusterStateFault``), defer to :class:`RedshiftClusterSettledTrigger`; the + ``_retry_delete_when_settled`` callback re-issues the delete once it settles. """ try: self.hook.delete_cluster( diff --git a/providers/amazon/src/airflow/providers/amazon/aws/triggers/redshift_cluster.py b/providers/amazon/src/airflow/providers/amazon/aws/triggers/redshift_cluster.py index 1f12e83192835..ce6d29154fb4d 100644 --- a/providers/amazon/src/airflow/providers/amazon/aws/triggers/redshift_cluster.py +++ b/providers/amazon/src/airflow/providers/amazon/aws/triggers/redshift_cluster.py @@ -353,10 +353,7 @@ class RedshiftClusterSettledTrigger(AwsBaseWaiterTrigger): 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 uses the custom ``cluster_deletable`` waiter, which has one - ``success`` acceptor per deletable ``ClusterStatus`` value. Any transitional state (``pausing``, - ``resizing``, ``resuming``, ``modifying``, ...) implicitly retries. A missing cluster - (``ClusterNotFound``) is treated as success since there is nothing left to wait on. The operator then - re-issues the delete. + ``success`` acceptor per deletable ``ClusterStatus`` value rather than a single target state. :param cluster_identifier: unique identifier of a cluster :param waiter_delay: The amount of time in seconds to wait between attempts. From 58f8d300b93dbbf2e7e4e8ce1e9faedd6589c0b2 Mon Sep 17 00:00:00 2001 From: Sean Ghaeli Date: Tue, 7 Jul 2026 21:42:01 +0000 Subject: [PATCH 3/7] Trim over-explanatory comments in redshift delete execute/init Generated-by: Claude Code (Opus) --- .../providers/amazon/aws/operators/redshift_cluster.py | 8 ++------ 1 file changed, 2 insertions(+), 6 deletions(-) diff --git a/providers/amazon/src/airflow/providers/amazon/aws/operators/redshift_cluster.py b/providers/amazon/src/airflow/providers/amazon/aws/operators/redshift_cluster.py index 9d993deca72f1..d1a1b9e54e571 100644 --- a/providers/amazon/src/airflow/providers/amazon/aws/operators/redshift_cluster.py +++ b/providers/amazon/src/airflow/providers/amazon/aws/operators/redshift_cluster.py @@ -867,9 +867,8 @@ def __init__( self.final_cluster_snapshot_identifier = final_cluster_snapshot_identifier self.wait_for_completion = wait_for_completion self.poll_interval = poll_interval - # Keep retrying while another operation is running on the cluster: a delete issued mid-transition - # (e.g. a pause/resize in progress) raises InvalidClusterStateFault. Retry until the cluster - # settles into a deletable state. 60 * 15s = ~15 min, long enough to outlast a cluster pause. + # Retry the delete while the cluster is mid-transition (InvalidClusterStateFault) until it + # settles into a deletable state. self._attempts = 60 self._attempt_interval = 15 self.deferrable = deferrable @@ -877,9 +876,6 @@ def __init__( def execute(self, context: Context): 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 From 10122f7674b1f8e8fdea4b688bd085fde68fb72f Mon Sep 17 00:00:00 2001 From: Sean Ghaeli Date: Tue, 7 Jul 2026 23:30:31 +0000 Subject: [PATCH 4/7] Add 'deletable' to spelling wordlist Generated-by: Claude Code (Opus) --- docs/spelling_wordlist.txt | 1 + 1 file changed, 1 insertion(+) diff --git a/docs/spelling_wordlist.txt b/docs/spelling_wordlist.txt index 2b7c13ee66036..ae56f4109449f 100644 --- a/docs/spelling_wordlist.txt +++ b/docs/spelling_wordlist.txt @@ -449,6 +449,7 @@ deferrable deidentify DeidentifyTemplate del +deletable delim deliverability deltalake From be576c9436e8a6d32bf6c466a2d869a20d1dc457 Mon Sep 17 00:00:00 2001 From: Sean Ghaeli Date: Mon, 20 Jul 2026 17:01:03 +0000 Subject: [PATCH 5/7] Use user-provided max_attempts for the sync busy-retry loop --- .../amazon/aws/operators/redshift_cluster.py | 23 ++++---- .../aws/operators/test_redshift_cluster.py | 52 +------------------ 2 files changed, 12 insertions(+), 63 deletions(-) diff --git a/providers/amazon/src/airflow/providers/amazon/aws/operators/redshift_cluster.py b/providers/amazon/src/airflow/providers/amazon/aws/operators/redshift_cluster.py index d1a1b9e54e571..d9ea42ce23479 100644 --- a/providers/amazon/src/airflow/providers/amazon/aws/operators/redshift_cluster.py +++ b/providers/amazon/src/airflow/providers/amazon/aws/operators/redshift_cluster.py @@ -837,9 +837,9 @@ class RedshiftDeleteClusterOperator(AwsBaseOperator[RedshiftHook]): https://boto3.amazonaws.com/v1/documentation/api/latest/reference/core/session.html :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: The maximum number of attempts to be made. In deferrable mode this bounds the - async wait for a busy cluster to settle before the delete is re-issued; combined with - ``poll_interval`` the default gives a ~15 minute window, long enough to outlast a pause/resize. + :param max_attempts: The maximum number of attempts to be made, both when retrying the delete while + the cluster is busy and when waiting for deletion to complete; combined with ``poll_interval`` + the default gives a ~15 minute window, long enough to outlast a pause/resize. """ template_fields: Sequence[str] = aws_template_fields( @@ -867,10 +867,6 @@ def __init__( self.final_cluster_snapshot_identifier = final_cluster_snapshot_identifier self.wait_for_completion = wait_for_completion self.poll_interval = poll_interval - # Retry the delete while the cluster is mid-transition (InvalidClusterStateFault) until it - # settles into a deletable state. - self._attempts = 60 - self._attempt_interval = 15 self.deferrable = deferrable self.max_attempts = max_attempts @@ -879,7 +875,10 @@ def execute(self, context: Context): self._delete_or_defer_until_settled() return - while self._attempts: + # Retry the delete while the cluster is mid-transition (InvalidClusterStateFault) until it + # settles into a deletable state. + attempts = self.max_attempts + while attempts: try: self.hook.delete_cluster( cluster_identifier=self.cluster_identifier, @@ -888,18 +887,18 @@ def execute(self, context: Context): ) break except self.hook.conn.exceptions.InvalidClusterStateFault: - self._attempts -= 1 + attempts -= 1 - if self._attempts: + if attempts: 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, ) - time.sleep(self._attempt_interval) + time.sleep(self.poll_interval) else: raise diff --git a/providers/amazon/tests/unit/amazon/aws/operators/test_redshift_cluster.py b/providers/amazon/tests/unit/amazon/aws/operators/test_redshift_cluster.py index c329b9431dd97..bcc854bfc40a4 100644 --- a/providers/amazon/tests/unit/amazon/aws/operators/test_redshift_cluster.py +++ b/providers/amazon/tests/unit/amazon/aws/operators/test_redshift_cluster.py @@ -782,63 +782,13 @@ def test_delete_cluster_multiple_attempts_fail(self, _, mock_conn, mock_delete_c cluster_identifier="test_cluster", aws_conn_id="aws_conn_test", wait_for_completion=False, + max_attempts=3, ) with pytest.raises(returned_exception): redshift_operator.execute(None) - assert mock_delete_cluster.call_count == 60 - - def test_busy_retry_defaults(self): - redshift_operator = RedshiftDeleteClusterOperator( - task_id="task_test", - cluster_identifier="test_cluster", - aws_conn_id="aws_conn_test", - ) - # 60 * 15s = ~15 min window, long enough to outlast a cluster pause/resize. - assert redshift_operator._attempts == 60 - assert redshift_operator._attempt_interval == 15 - - @mock.patch.object(RedshiftHook, "delete_cluster") - @mock.patch.object(RedshiftHook, "conn") - @mock.patch("time.sleep", return_value=None) - def test_delete_cluster_exhausts_busy_retries_then_raises(self, _, mock_conn, mock_delete_cluster): - exception = boto3.client("redshift").exceptions.InvalidClusterStateFault({}, "test") - returned_exception = type(exception) - mock_conn.exceptions.InvalidClusterStateFault = returned_exception - mock_delete_cluster.side_effect = exception - - redshift_operator = RedshiftDeleteClusterOperator( - task_id="task_test", - cluster_identifier="test_cluster", - aws_conn_id="aws_conn_test", - wait_for_completion=False, - ) - redshift_operator._attempts = 3 - with pytest.raises(returned_exception): - redshift_operator.execute(None) - assert mock_delete_cluster.call_count == 3 - @mock.patch.object(RedshiftHook, "delete_cluster") - @mock.patch.object(RedshiftHook, "conn") - @mock.patch("time.sleep", return_value=None) - def test_delete_cluster_succeeds_on_second_attempt(self, _, mock_conn, mock_delete_cluster): - exception = boto3.client("redshift").exceptions.InvalidClusterStateFault({}, "test") - returned_exception = type(exception) - mock_conn.exceptions.InvalidClusterStateFault = returned_exception - mock_delete_cluster.side_effect = [exception, True] - - redshift_operator = RedshiftDeleteClusterOperator( - task_id="task_test", - cluster_identifier="test_cluster", - aws_conn_id="aws_conn_test", - wait_for_completion=False, - ) - redshift_operator._attempt_interval = 0 - redshift_operator.execute(None) - - assert mock_delete_cluster.call_count == 2 - @mock.patch("airflow.providers.amazon.aws.hooks.redshift_cluster.RedshiftHook.cluster_status") @mock.patch("airflow.providers.amazon.aws.hooks.redshift_cluster.RedshiftHook.delete_cluster") def test_delete_cluster_deferrable_mode(self, mock_delete_cluster, mock_cluster_status): From 29c6459dd932119c2564973d71b74fa67f265ac7 Mon Sep 17 00:00:00 2001 From: Sean Ghaeli Date: Mon, 20 Jul 2026 22:25:37 +0000 Subject: [PATCH 6/7] Restore cluster_identifier from trigger event on deferral resume --- .../amazon/aws/operators/redshift_cluster.py | 1 + .../amazon/aws/triggers/redshift_cluster.py | 3 +- .../aws/operators/test_redshift_cluster.py | 29 +++++++++++++++++-- 3 files changed, 30 insertions(+), 3 deletions(-) diff --git a/providers/amazon/src/airflow/providers/amazon/aws/operators/redshift_cluster.py b/providers/amazon/src/airflow/providers/amazon/aws/operators/redshift_cluster.py index d9ea42ce23479..2424e35e0ca1f 100644 --- a/providers/amazon/src/airflow/providers/amazon/aws/operators/redshift_cluster.py +++ b/providers/amazon/src/airflow/providers/amazon/aws/operators/redshift_cluster.py @@ -976,6 +976,7 @@ def _retry_delete_when_settled(self, context: Context, event: dict[str, Any] | N if validated_event["status"] != "success": raise AirflowException(f"Error waiting for cluster to become deletable: {validated_event}") + self.cluster_identifier = validated_event["cluster_identifier"] self._delete_or_defer_until_settled() def execute_complete(self, context: Context, event: dict[str, Any] | None = None) -> None: diff --git a/providers/amazon/src/airflow/providers/amazon/aws/triggers/redshift_cluster.py b/providers/amazon/src/airflow/providers/amazon/aws/triggers/redshift_cluster.py index ce6d29154fb4d..8f43ef2e4f757 100644 --- a/providers/amazon/src/airflow/providers/amazon/aws/triggers/redshift_cluster.py +++ b/providers/amazon/src/airflow/providers/amazon/aws/triggers/redshift_cluster.py @@ -381,7 +381,8 @@ def __init__( failure_message="Error while waiting for the redshift cluster to become deletable", status_message="Waiting for redshift cluster to settle into a deletable state", status_queries=["Clusters[].ClusterStatus"], - return_value=None, + return_key="cluster_identifier", + return_value=cluster_identifier, waiter_delay=waiter_delay, waiter_max_attempts=waiter_max_attempts, aws_conn_id=aws_conn_id, diff --git a/providers/amazon/tests/unit/amazon/aws/operators/test_redshift_cluster.py b/providers/amazon/tests/unit/amazon/aws/operators/test_redshift_cluster.py index bcc854bfc40a4..46d25b8ab33b4 100644 --- a/providers/amazon/tests/unit/amazon/aws/operators/test_redshift_cluster.py +++ b/providers/amazon/tests/unit/amazon/aws/operators/test_redshift_cluster.py @@ -870,12 +870,37 @@ def test_retry_delete_when_settled_reissues_delete(self, mock_delete_cluster, mo with pytest.raises(TaskDeferred) as exc: delete_cluster._retry_delete_when_settled( - context=None, event={"status": "success", "message": "Cluster settled"} + context=None, event={"status": "success", "cluster_identifier": "test_cluster"} ) assert isinstance(exc.value.trigger, RedshiftDeleteClusterTrigger) mock_delete_cluster.assert_called_once() + @mock.patch("airflow.providers.amazon.aws.hooks.redshift_cluster.RedshiftHook.cluster_status") + @mock.patch("airflow.providers.amazon.aws.hooks.redshift_cluster.RedshiftHook.delete_cluster") + def test_retry_delete_when_settled_uses_cluster_id_from_event( + self, mock_delete_cluster, mock_cluster_status + ): + """The re-issued delete targets the identifier from the trigger event, not the operator field.""" + mock_delete_cluster.return_value = True + mock_cluster_status.return_value = "cluster_not_found" + delete_cluster = RedshiftDeleteClusterOperator( + task_id="task_test", + cluster_identifier="rerendered_different_value", + deferrable=True, + wait_for_completion=False, + ) + + delete_cluster._retry_delete_when_settled( + context=None, event={"status": "success", "cluster_identifier": "original_cluster"} + ) + + mock_delete_cluster.assert_called_once_with( + cluster_identifier="original_cluster", + skip_final_cluster_snapshot=True, + final_cluster_snapshot_identifier=None, + ) + @mock.patch.object(RedshiftHook, "conn") @mock.patch("airflow.providers.amazon.aws.hooks.redshift_cluster.RedshiftHook.delete_cluster") def test_retry_delete_when_settled_redefers_on_race(self, mock_delete_cluster, mock_conn): @@ -893,7 +918,7 @@ def test_retry_delete_when_settled_redefers_on_race(self, mock_delete_cluster, m with pytest.raises(TaskDeferred) as exc: delete_cluster._retry_delete_when_settled( - context=None, event={"status": "success", "message": "Cluster settled"} + context=None, event={"status": "success", "cluster_identifier": "test_cluster"} ) assert isinstance(exc.value.trigger, RedshiftClusterSettledTrigger) From e64670f4afcde6fb32655e01ee7a6527a7039f22 Mon Sep 17 00:00:00 2001 From: Sean Ghaeli Date: Mon, 20 Jul 2026 23:13:27 +0000 Subject: [PATCH 7/7] Retrigger CI (unrelated static-check flake on migration references)