apache / apache/airflow

`wait_for_past_depends_before_skipping` does not seem to have any effect

Open
#55,146 5 comments 0 reactions 0 assignees View on GitHub
area:core kind:bug needs-triage
Dominant language
Python
Stars
46.9k
Forks
17.8k
Avg merge
2d 7h
Merged PRs (30d)
484

Description

### Apache Airflow version

Airflow 3.1.0

### What happened?

I am part of a data engineering team. We use airflow to orchestrate our data ingestion pipelines. At a high-level, they are made of:

- An ingestion task group, that moves data from a source system to our S3 storage
- Validation tests
- Processing jobs that clean up and prepares the data for use by downstream consumers
- Validations tests on the processed data

The processing jobs run in Spark. To avoid overhead, I have been looking at using Airflow's skipping mechanisms to skip the rest of the pipeline if there is no data to be ingested.

However, we need the processing jobs to run sequentially. For this purpose, we use `depends_on_past`. I then stumbled upon [the PR](https://github.com/apache/airflow/pull/27710) introducing `wait_for_past_depends_before_skipping`, which seems like it would allow me to skip tasks while still preserving the dependencies on past runs. However, the flag does not seem to have an effect.

### What you think should happen instead?

The flag should behave as advertised in the base PR and in the docs.

### How to reproduce

```python
from airflow import DAG
from airflow.models import Param
from airflow.providers.standard.operators.empty import EmptyOperator
from airflow.providers.standard.operators.python import ShortCircuitOperator
from airflow.sdk import get_current_context

def check_continue():
ctx = get_current_context()
skip_downstream = ctx["params"]["skip"]
return not skip_downstream

with DAG(
dag_id="test_airflow_skip",
schedule=None,
params={
"skip": Param(default=False, type="boolean"),
},
) as dag:
depends_on_past = EmptyOperator(
task_id="depends_on_past",
depends_on_past=True,
wait_for_past_depends_before_skipping=True,
)
skip = ShortCircuitOperator(
task_id="skip",
python_callable=check_continue,
)

skip >> depends_on_past
```
1. Launch a run
2. Then manually set the state of the `depends_on_past` task to `failed`
3. Then launch a run with the `skip` parameter enabled: `depends_on_past` will be skipped when it shouldn't be.

Image

### Operating System

Ubuntu 22.04.5 LTS

### Versions of Apache Airflow Providers

_No response_

### Deployment

Docker-Compose

### Deployment details

_No response_

### Anything else?

_No response_

### Are you willing to submit PR?

- [ ] Yes I am willing to submit a PR!

### Code of Conduct

- [x] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)

Contributor guide

Open the contributing guide

Research direction

Reproduce the behavior with the provided DAG, EmptyOperator, ShortCircuitOperator, and depends_on_past settings, then compare the flag's implementation with PR 27710 and the referenced documentation. Done means a failed previous task prevents the current task from being skipped when wait_for_past_depends_before_skipping is enabled, with coverage for the supplied sequence.

Written by the indexing model from the issue text.

Assessment

Tech stack
docker-compose, python
Domain
backend, data-engineering
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
42/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.