vincbeck commented on code in PR #63074:
URL: https://github.com/apache/airflow/pull/63074#discussion_r2906122582
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/redshift_cluster.py:
##########
@@ -235,6 +239,70 @@ def __init__(
self.deferrable = deferrable
self.kwargs = kwargs
self.delete_cluster_on_failure = delete_cluster_on_failure
+ self.cleanup_timeout_seconds = cleanup_timeout_seconds
+
+ def _attempt_cleanup_with_retry(self) -> None:
+ """
+ Attempt bounded best-effort deletion of the cluster.
+
+ This method is only invoked during task failure handling.
+ It does not block until deletion completes and will not
+ mask the original exception.
+ """
+ RETRY_INTERVAL_SECONDS = 60
+
+ # Bound cleanup attempts to avoid indefinitely occupying a worker slot.
+ deadline = time.monotonic() + self.cleanup_timeout_seconds
+ attempt = 1
+
+ while True:
+ try:
+ self.log.info(
+ "Attempt %s: Deleting Redshift cluster %s.",
+ attempt,
+ self.cluster_identifier,
+ )
+
+ # Do not wait for deletion to complete; cleanup is best-effort.
+
self.hook.delete_cluster(cluster_identifier=self.cluster_identifier)
Review Comment:
You are basically calling `self.hook.delete_cluster` with some retry
strategy. Could you use `tenacity`? It would make the code way cleaner.
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/redshift_cluster.py:
##########
@@ -193,6 +196,7 @@ def __init__(
poll_interval: int = 60,
deferrable: bool = conf.getboolean("operators", "default_deferrable",
fallback=False),
delete_cluster_on_failure: bool = True,
+ cleanup_timeout_seconds: int = 300,
Review Comment:
The default value does not match the docstring
--
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]
