seanghaeli commented on code in PR #69574:
URL: https://github.com/apache/airflow/pull/69574#discussion_r3545689248
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/redshift_cluster.py:
##########
@@ -897,36 +903,82 @@ 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), 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._delete_or_defer_until_settled()
Review Comment:
I believe the exact same pattern is used here in EKS:
https://github.com/apache/airflow/blob/e5465795a887129c738c7c5d1e5f0a7b4bc421a8/providers/amazon/src/airflow/providers/amazon/aws/operators/eks.py#L401
I am going to investigate if this causes a fail but I think if this is an
issue in this PR, it's also a latent issue in EKS currently
--
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]
