seanghaeli commented on code in PR #69574:
URL: https://github.com/apache/airflow/pull/69574#discussion_r3548059207
##########
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:
@vincbeck I just tried the resume step from a different worker than the one
that tries the first delete (which fails because the cluster is not in a ready
state yet), and it still succeeds.
1. create redshift cluster
2. paused the cluster then resumed it (to put it in a mid-transition state)
3. triggered `RedshiftDeleteClusterOperator` in deferrable mode
4. What I observed: worker process `pid 738` ran `execute` while cluster was
still mid-transition so it defers.
5. later, a different worker `pid 968` successfully deleted the cluster.
I believe this is fine because the `cluster_identifier` is serialized in the
trigger's init, so it persists in the metastore when the new worker reads it:
https://github.com/apache/airflow/blob/e5465795a887129c738c7c5d1e5f0a7b4bc421a8/task-sdk/src/airflow/sdk/execution_time/task_runner.py#L1453-L1457
^ where `cluster_identifier` is included in `trigger_kwargs`
--
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]
