udsy19 opened a new pull request, #73470:
URL: https://github.com/apache/airflow/pull/73470
`wait_for_past_depends_before_skipping` is documented to hold a
`depends_on_past` task back — leave it in `none` state — instead of skipping
it, whenever its own past-run dependency is not yet met. That contract is
honored in exactly one place: `NotPreviouslySkippedDep`, the scheduler
dependency check that re-evaluates a task recorded in a `SkipMixin` XCom
payload and checks the `PAST_DEPENDS_MET` XCom before letting it be skipped.
The problem: `SkipMixin.skip()` (used by `ShortCircuitOperator`) and
`SkipMixin.skip_all_except()` (used by `BranchPythonOperator`) never route
through that check for their own direct targets. Both raise
`DownstreamTasksSkipped`, which the task-sdk supervisor turns into a `PATCH
.../skip-downstream` call
(`airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py::ti_skip_downstream`)
that force-writes `state=SKIPPED` on the named task instances directly — no
`depends_on_past`/`wait_for_past_depends_before_skipping` check anywhere on
that path, only a guard against clobbering a RUNNING/SUCCESS/FAILED row. So a
task with the flag set gets skipped immediately by its upstream
short-circuit/branch, regardless of whether its own past run succeeded.
Impact: contract-violation
WHO reaches this / entry point: any DAG using the documented
`wait_for_past_depends_before_skipping=True` + `depends_on_past=True`
combination on a task that sits downstream of a `ShortCircuitOperator` or
`BranchPythonOperator` — exactly the workflow the flag's own PR (#27710) was
written for, and exactly apache/airflow#55146's own reproduction. Triggered by:
the upstream `ShortCircuitOperator`/`BranchPythonOperator` deciding to skip
that specific downstream task in the current DAG run (short-circuit condition
false, or branch not taken). What they observe: the task is marked `SKIPPED`
immediately, even when the previous DAG run's instance of that same task failed
— `wait_for_past_depends_before_skipping` has no effect at all for this (very
common) skip path, as reported.
## Fix
`SkipMixin.skip()`/`skip_all_except()` already push the `XCOM_SKIPMIXIN_KEY`
payload listing every affected task before deciding what to force-skip — that
payload is what `NotPreviouslySkippedDep` reads when it later re-evaluates a
task. Mapped tasks are already excluded from the immediate force-skip and left
for that same check ("future mapped tasks have not been expanded yet and are
handled by NotPreviouslySkippedDep"). This PR extends the same exclusion to any
task with `wait_for_past_depends_before_skipping=True`: it is still recorded in
the XCom payload (so `NotPreviouslySkippedDep` knows it should be skipped once
its past depends are met), but it is no longer force-skipped by the direct
`skip-downstream` write. `NotPreviouslySkippedDep` picks it up on the next
scheduler pass and correctly holds it in `none` until `PAST_DEPENDS_MET` is
true.
## Testing
Two new tests in
`task-sdk/tests/task_sdk/bases/test_skipmixin.py::TestSkipMixin`:
- `test_skip_excludes_wait_for_past_depends_task` (the
`ShortCircuitOperator` path via `skip()`)
- `test_skip_all_except_excludes_wait_for_past_depends_task` (the
`BranchPythonOperator` path via `skip_all_except()`)
Negative control — reverting only
`task-sdk/src/airflow/sdk/bases/skipmixin.py` to `upstream/main` and running
the full test file:
```
FAILED
task-sdk/tests/task_sdk/bases/test_skipmixin.py::TestSkipMixin::test_skip_excludes_wait_for_past_depends_task
FAILED
task-sdk/tests/task_sdk/bases/test_skipmixin.py::TestSkipMixin::test_skip_all_except_excludes_wait_for_past_depends_task
2 failed, 12 passed, 1 warning in 2.48s
```
With the fix restored, the full file passes: `14 passed, 1 warning`. The
other 12 pre-existing tests in the file are unmodified in behavior — they now
pass an explicit `wait_for_past_depends_before_skipping=False` on their mock
task fixtures, because without it the new filtering logic reads an
auto-generated (truthy) `MagicMock` attribute from bare
`MagicMock(spec=BaseOperator)` instances and incorrectly defers them too. That
is a mocking artifact, not a production behavior change: real
`BaseOperator`/`MappedOperator` instances always carry the real
dataclass-defaulted `False`.
closes: #55146
Signed-off-by: Udaya Tejas <[email protected]>
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes - Claude Code (Sonnet 5)
Generated-by: Claude Code (Sonnet 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]
