Make RedshiftDeleteClusterOperator delete reliably during cluster transitions

[seanghaeli]

Why

RedshiftDeleteClusterOperator retries on InvalidClusterStateFault when the cluster is mid-transition (e.g. pausing/resuming/resizing), but the retry budget was hardcoded to 10 attempts * 15s = 2.5 minutes. A cluster transition routinely takes much longer than that (a pause is ~2–15 min, a classic resize ~6 min), so the loop exhausts and re-raises long before the cluster reaches a deletable state. The task fails and the cluster keeps running until manual deletion (a leak).

Example errors seen mid-transition:

InvalidClusterState: There is an operation running on the Cluster.
Please try to delete it at a later time.

Unable to delete the cluster ... You can only delete clusters with PAUSED, ..., ACTIVE, ... lifecycles.

What

Two changes, one per mode:

1. Non-deferrable mode — configurable, longer retry budget. Promote the two hardcoded constants to __init__ params and raise the default so the retry outlasts a real transition:

• busy_retry_attempts: int = 60
• busy_retry_interval: int = 15

Default is now 60 * 15s = 15 minutes, both overridable. Fail-loud is preserved (re-raises once the budget is exhausted). The private _attempts / _attempt_interval attributes are replaced by these params.

2. Deferrable mode — async re-defer instead of a synchronous block. Previously the synchronous retry loop ran in deferrable mode too, blocking the worker (this was flagged by @ramitkataria on an earlier iteration — a 15-min sync block defeats the purpose of deferrable mode). Deferrable mode now:

• attempts the delete once; on InvalidClusterStateFault it defers to a new RedshiftClusterSettledTrigger that polls until the cluster is no longer in a transitional state,
• then a callback re-issues the delete and defers to the existing RedshiftDeleteClusterTrigger (waits cluster_deleted),
• if the re-issued delete races another transition, it re-defers to the settle trigger.

No worker is blocked at any point. This mirrors the re-defer pattern in EksCreateClusterOperator.deferrable_create_cluster_next.

Testing

Unit tests (added/updated): configurable defaults/custom values, retry-then-raise, succeed-on-second-attempt (sync); defer-to-settle-trigger, callback re-issues delete, re-defer-on-race, error-event-raises (deferrable). All pass.

End-to-end before/after — non-deferrable mode (real AWS)

Identical DAG, identical ~6.5-minute transient (classic resize), controlled before/after against a real cluster.

BEFORE (stock main, 2.5-min budget) — task FAILS, cluster leaks:

20:38:51  Cluster in resizing state, unable to delete. 9 attempts remaining.
   ... (15s sleeps) ...
20:40:54  ... 1 attempts remaining.
botocore.errorfactory.InvalidClusterStateFault: ... Unable to delete the cluster ...
DagRun ... state=failed   TASK_EXIT=1

AFTER (this PR, 15-min budget) — task SUCCEEDS, cluster deleted:

20:42:27  Cluster in resizing state, unable to delete. 59 attempts remaining.
20:43:29  ... 55 attempts remaining.      <-- already past where stock main died
20:44:46  ... 50 attempts remaining.
20:45:01  Task instance state updated ... new_state=success   <-- delete accepted once resize settled
DagRun ... state=success   TASK_EXIT=0

End-to-end — deferrable mode async re-defer (real AWS)

Same real-cluster + classic-resize setup, deferrable=True. The operator deferred through the whole transient without blocking a worker:

23:04:52  Cluster rs-e2e-deferrable is busy; deferring until it settles into a deletable state.
23:04:52  Pausing task as DEFERRED.
23:04:52  [DAG TEST] running trigger in line     <-- RedshiftClusterSettledTrigger polls through the resize
   ... ~5.5 min real-time wait ...
23:10:15  [DAG TEST] Trigger completed           <-- cluster settled (available); callback re-issues delete
   ... (cluster briefly re-entered a transitional state; callback re-deferred to settle trigger 3x — race path exercised live) ...
23:10:17  [DAG TEST] running trigger in line     <-- RedshiftDeleteClusterTrigger waits cluster_deleted
23:10:17  Redshift Cluster deletion in progress: ['deleting']

AWS-side timeline confirms: resizing → available → deleting → ClusterNotFound. Cluster fully deleted, no leak. (A live race — the cluster momentarily bounced back to a transitional state right after settling — was handled correctly by the re-defer branch.)

Note: task logs are the verification artifact (no UI screenshots; headless). A classic resize was used as the transient because pause/resume settled too quickly (~100–170s) to reliably exceed stock main's 150s budget — the transition type is immaterial; any InvalidClusterStateFault-producing operation exercises the same path.

---

Generated-by: Claude Code (Opus)