PradeepNalluri opened a new pull request, #73935:
URL: https://github.com/apache/airflow/pull/73935
`MwaaServerlessStartWorkflowRunOperator` returned as soon as a workflow run
was created, so waiting for the run needed a separate
`MwaaServerlessWorkflowRunSensor`, which could only poke and held a worker slot
for the whole run. Other AWS operators (Glue, EMR Serverless, classic MWAA) can
wait for or defer on the work they start.
This adds:
- `wait_for_completion`, `deferrable`, `waiter_delay` and
`waiter_max_attempts` to `MwaaServerlessStartWorkflowRunOperator`.
`wait_for_completion` defaults to `False`, so existing Dags behave the same.
- `client_token` to `MwaaServerlessStartWorkflowRunOperator`, passed to
`StartWorkflowRun` as `ClientToken`, so a retried task returns the existing run
instead of starting a duplicate.
- `deferrable` and `max_retries` to `MwaaServerlessWorkflowRunSensor`.
- A `workflow_run_complete` waiter for `mwaa-serverless` and a new
`MwaaServerlessWorkflowRunCompletedTrigger`, which supports custom success and
failure states.
**Behaviour change in the existing sensor:** when `success_states` includes
a default failure state (for example `{"SUCCESS", "STOPPED"}`) and
`failure_states` is not set, the default failure states now exclude the states
listed in `success_states`. Previously, the sensor failed a stopped run even
though `STOPPED` was listed as a success.
### Testing
- Unit tests for the operator, sensor, trigger and waiter. Every new test
fails without the source change.
- The MWAA, MWAA Serverless, AWS waiter, AWS trigger base and template-field
test suites pass.
- mypy on the changed modules, and the prek pre-commit hooks.
- End to end against a real MWAA Serverless workflow in a test account,
calling the operators directly: synchronous wait, deferral with the trigger
polling through aiobotocore, `client_token` returning the same run, and a
stopped run failing the sensor.
- End to end in `breeze start-airflow` with the scheduler and triggerer:
deferred operator and sensor tasks resumed and succeeded, the sensor with
default states failed on a stopped run, and the sensor with `STOPPED` in
`success_states` succeeded in both deferrable and poke mode.
---
##### Was generative AI tooling used to co-author this PR?
- [x] Yes — Claude Code (Opus 5.5)
Generated-by: Claude Code (Opus 5.5) following [the
guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
--
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]
