This is an automated email from the ASF dual-hosted git repository.
vincbeck pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 1fd77cdd428 Make RedshiftDeleteClusterOperator delete reliably during
cluster transitions (#69574)
1fd77cdd428 is described below
commit 1fd77cdd4282a53be1103c15c1f8976f03831b45
Author: Sean Ghaeli <[email protected]>
AuthorDate: Wed Jul 22 12:44:58 2026 -0700
Make RedshiftDeleteClusterOperator delete reliably during cluster
transitions (#69574)
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.
---
docs/spelling_wordlist.txt | 1 +
.../amazon/aws/operators/redshift_cluster.py | 122 +++++++++++-----
.../amazon/aws/triggers/redshift_cluster.py | 54 ++++++++
.../providers/amazon/aws/waiters/redshift.json | 55 ++++++++
.../amazon/aws/operators/test_redshift_cluster.py | 153 +++++++++++++++++++--
.../amazon/aws/triggers/test_redshift_cluster.py | 36 ++++-
6 files changed, 369 insertions(+), 52 deletions(-)
diff --git a/docs/spelling_wordlist.txt b/docs/spelling_wordlist.txt
index cf74e1f5f07..2a8b7c94591 100644
--- a/docs/spelling_wordlist.txt
+++ b/docs/spelling_wordlist.txt
@@ -452,6 +452,7 @@ deferrable
deidentify
DeidentifyTemplate
del
+deletable
DeleteAgentResponse
DeleteAgentVersionResponse
delim
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 ea0d1272db8..2424e35e0ca 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 tenacity import Retrying, retry_if_exception,
stop_after_delay, wait_fixed
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, 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(
@@ -864,16 +867,18 @@ class
RedshiftDeleteClusterOperator(AwsBaseOperator[RedshiftHook]):
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
self.deferrable = deferrable
self.max_attempts = max_attempts
def execute(self, context: Context):
- while self._attempts:
+ if self.deferrable:
+ self._delete_or_defer_until_settled()
+ return
+
+ # 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,
@@ -882,51 +887,98 @@ class
RedshiftDeleteClusterOperator(AwsBaseOperator[RedshiftHook]):
)
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
- 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), then defer.
+
+ 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(
+ 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.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:
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 ebd32a2b423..8f43ef2e4f7 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,57 @@ class RedshiftClusterTrigger(BaseTrigger):
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 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.
+ :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_key="cluster_identifier",
+ return_value=cluster_identifier,
+ 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 8165eb3fc43..389da67068e 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 db6466125a0..46d25b8ab33 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 @@ from airflow.providers.amazon.aws.operators.redshift_cluster
import (
RedshiftResumeClusterOperator,
)
from airflow.providers.amazon.aws.triggers.redshift_cluster import (
+ RedshiftClusterSettledTrigger,
RedshiftCreateClusterSnapshotTrigger,
RedshiftDeleteClusterTrigger,
RedshiftPauseClusterTrigger,
@@ -781,18 +782,19 @@ class TestDeleteClusterOperator:
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 == 10
+ assert mock_delete_cluster.call_count == 3
@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 +808,59 @@ class TestDeleteClusterOperator:
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",
@@ -823,10 +868,98 @@ class TestDeleteClusterOperator:
wait_for_completion=False,
)
+ with pytest.raises(TaskDeferred) as exc:
+ delete_cluster._retry_delete_when_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):
+ """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",
+ deferrable=True,
+ wait_for_completion=False,
+ )
+
+ with pytest.raises(TaskDeferred) as exc:
+ delete_cluster._retry_delete_when_settled(
+ context=None, event={"status": "success",
"cluster_identifier": "test_cluster"}
+ )
+
+ 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 a55494ce8ab..42f0b5f1f80 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 @@ from unittest import mock
import pytest
from airflow.providers.amazon.aws.triggers.redshift_cluster import (
+ RedshiftClusterSettledTrigger,
RedshiftClusterTrigger,
RedshiftCreateClusterSnapshotTrigger,
RedshiftCreateClusterTrigger,
@@ -152,43 +153,64 @@ WAITER_TRIGGER_PARAMS = [
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 @@ class TestRedshiftWaiterTriggers:
}
@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 @@ class TestRedshiftWaiterTriggers:
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",
