saadbelgi opened a new pull request, #71652: URL: https://github.com/apache/airflow/pull/71652
Add deferrable mode to `EmrServerlessJobSensor` so waiting on a long-running EMR Serverless job does not occupy a worker slot. Follows `EmrJobFlowSensor` / `EmrStepSensor`: poke once, then `defer()` to the existing `EmrServerlessStartJobTrigger` with `cancel_on_kill=False` (this sensor did not start the job). Poke/reschedule behaviour is unchanged when `deferrable=False`. Docs: class docstring plus `providers/amazon/docs/operators/emr/emr_serverless.rst`. ### Known limitations These match the `EmrJobFlowSensor`/`EmrStepSensor` unless noted. - **`target_states` after deferral.** Poke mode honours `target_states` (default `SUCCESS`). After deferral the waiter is `serverless_job_completed` (`SUCCESS` / `FAILED` / `CANCELLED`). The first poke still uses `target_states`, so a Dag waiting for `RUNNING` can succeed without deferring if the job is already running; otherwise it waits for completion. - **`max_attempts` only applies when `deferrable=True`.** Poke/reschedule uses `timeout` (default `sensors.default_timeout`, 7 days). Deferrable mode uses `max_attempts` (default 60) as waiter polls and as `defer(timeout=max_attempts * poke_interval)` (~1 hour at the default 60s `poke_interval`). - **`execute_complete` raises `RuntimeError`** as new `raise AirflowException` is blocked by `check-no-new-airflow-exceptions` pre-commit hook. JobFlow/Step still raise `AirflowException`. `poke()` still raises `AirflowException` on `JOB_FAILURE_STATES`. Do let me know if this is inconsistent or if a specific new exception class would be required. - I am reusing `EmrServerlessStartJobTrigger` (used in `EmrServerlessStartJobOperator`) without renaming it, as it already implements the functionality needed and I wanted to keep the change-set as small as possible for my first contribution. Do let me know if this is not the correct thing to do. --- ##### Was generative AI tooling used to co-author this PR? - [X] Yes — Cursor Grok 4.6 Generated-by: Cursor Grok 4.6 following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions) --- ### Tests run locally - **prek (commit):** ruff, ruff-format, and `check-no-new-airflow-exceptions` on the three changed files. - **mypy:** `breeze run mypy providers/amazon/src/airflow/providers/amazon/aws/sensors/emr.py` - **Sensor unit tests:** `breeze testing providers-tests --python 3.10 providers/amazon/tests/unit/amazon/aws/sensors/test_emr_serverless_job.py -xvs` — 13 passed - **Sensor + EMR triggers:** same command plus `providers/amazon/tests/unit/amazon/aws/triggers/test_emr.py` — 36 passed - **Amazon provider suite:** `breeze testing providers-tests --python 3.10 --test-type "Providers[amazon]"` — 4734 passed, 9 skipped --- -- 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]
