Make RedshiftDeleteClusterOperator delete reliably during cluster transitions

[seanghaeli]

Why

RedshiftDeleteClusterOperator retries on InvalidClusterStateFault when the cluster is mid-transition (pausing/resuming/resizing), but the retry budget was hardcoded to 10 attempts * 15s = 2.5 minutes. Transitions routinely take longer, so the loop exhausts and re-raises before the cluster becomes deletable.

The error seen mid-transition:

InvalidClusterState: Unable to delete the cluster ... You can only delete clusters with
ACTIVE, INCOMPATIBLE_HSM, INSUFFICIENT_VPCE, HARDWARE_FAILURE, INCOMPATIBLE_NETWORK,
PAUSED, INACCESSIBLE_KMS_KEY, INSUFFICIENT_CAPACITY, INCOMPATIBLE_RESTORE lifecycles.

What

Non-deferrable mode: raise the synchronous busy-retry budget to 60 * 15s = ~15 min so it outlasts a transition. Still throw failure once timed out.

Deferrable mode: previously the synchronous retry loop ran before the defer, blocking the worker for the whole budget (defeating the purpose of deferrable mode). Now the operator attempts the delete once; on InvalidClusterStateFault it defers to a new RedshiftClusterSettledTrigger that treats transitional states as retry and fires once the cluster reaches a deletable lifecycle. A callback then re-issues the delete and defers to the existing RedshiftDeleteClusterTrigger (cluster_deleted); on a race it re-defers to the settle trigger. No worker is blocked at any point.

Mirrors the deferrable re-defer pattern established for EKS in #32355

Testing

End-to-end before/after

Non-deferrable — BEFORE: FAILS, cluster leaks:

20:32:04  Cluster in resizing state, unable to delete. 9 attempts remaining.
   ... (15s apart, 9→1) ...
20:34:22  botocore.errorfactory.InvalidClusterStateFault: Unable to delete the cluster rs-e2e-sync ...
          DagRun ... state=failed   TASK_EXIT=1

Non-deferrable — AFTER: SUCCEEDS, cluster deleted:

20:58:33  Cluster in resizing state, unable to delete. 49 attempts remaining.   <-- past stock's 10-cap
21:01:07  Cluster in resizing state, unable to delete. 39 attempts remaining.
21:01:22  new_state=success   DagRun ... state=success   TASK_EXIT=0

Deferrable — BEFORE: blocks worker 2.5 min then FAILS (never defers):

20:51:53  Cluster in resizing state, unable to delete. 9 attempts remaining.
   ... 9→1 ...
20:54:11  new_state=failed   DagRun ... state=failed   TASK_EXIT=1

Deferrable — AFTER: async wait via cluster_deletable waiter, then delete. SUCCEEDS:

21:02:10  Cluster rs-e2e-async is busy; deferring until it settles into a deletable state.
21:02:10  Pausing task as DEFERRED.
21:02:11  Waiting for redshift cluster to settle into a deletable state: ['resizing']
   ... waiter treats 'resizing' as retry for ~10 min, task stays DEFERRED (worker not blocked) ...
21:12:38  (settled) delete re-issued and accepted -> defer to cluster_deleted
21:15:10  Redshift Cluster deletion in progress: ['deleting']
21:16:10  Cluster deleted successfully   DagRun ... state=success   TASK_EXIT=0

[vincbeck]

This might not work if _retry_delete_when_settled is executed on a different machine/host than execute. If some random values are passed to the construct of RedshiftDeleteClusterOperator, then some values will differ between execute and _retry_delete_when_settled. For example, if a random number is used to define cluster_identifier, then definitely this will fail

[seanghaeli]

I believe the exact same pattern is used here in EKS:

airflow/providers/amazon/src/airflow/providers/amazon/aws/operators/eks.py

Line 401 in e546579

self.hook.delete_cluster(name=self.cluster_name)

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