HelmiDev03 opened a new pull request, #73946: URL: https://github.com/apache/airflow/pull/73946
In deferrable mode, `EmrCreateJobFlowOperator` ignores `wait_policy=WaitPolicy.WAIT_FOR_STEPS_COMPLETION`: the task is marked successful as soon as the EMR cluster is running, while its steps are still executing, and a failure after that point goes unnoticed. The non-deferrable path correctly waits for the cluster to terminate. #61195 made the operator pass the waiter selected from `wait_policy` (`job_flow_terminated` for `WAIT_FOR_STEPS_COMPLETION`) to `EmrCreateJobFlowTrigger`, but the trigger never added `waiter_name` to its `serialized_fields`. The triggerer rebuilds the trigger from `serialize()`, so it always fell back to the default `job_flow_waiting` waiter, which already succeeds on `RUNNING`. This adds `waiter_name` to the trigger's serialized kwargs. Triggers deferred before this change have no `waiter_name` in their stored kwargs and keep rebuilding with the default, so their behaviour does not change. Tests: - `TestEmrCreateJobFlowTrigger.test_serialization` pinned the serialized kwargs without `waiter_name`; it now covers the default and `job_flow_terminated` waiters and checks that a trigger rebuilt from the serialized kwargs keeps the waiter. - New `test_create_job_flow_deferrable_serializes_wait_policy_waiter` checks, for each `WaitPolicy`, that the deferred trigger's serialized kwargs carry the matching waiter. Both fail without the fix (kwargs mismatch / `KeyError: 'waiter_name'`) and pass with it. Also ran locally: `triggers/test_emr.py`, `triggers/test_serialization.py`, `operators/test_emr_create_job_flow.py` and `test_deferred_hook_configuration.py` (295 passed, 35 skipped by that test's allowlist), `ruff check` / `ruff format --check`, `scripts/ci/prek/check_trigger_serialize_init.py`, and `mypy` on the changed module. The tests ran on Windows with local stubs for `os.register_at_fork` and `fcntl`, which Airflow needs at import time; CI covers Linux. related: #61195 --- ##### 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) --- * Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)** for more information. Note: commit author/co-author name and email in commits become permanently public when merged. * For fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed. * When adding dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x). * For significant user-facing changes create newsfragment: `{pr_number}.significant.rst`, in [airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/newsfragments). You can add this file in a follow-up commit after the PR is created so you know the PR number. -- 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]
