You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "shai-ikko (via GitHub)" <gi...@apache.org> on 2024/07/17 15:16:17 UTC
[I] Task with "wait_for_downstream" doesn't wait, because start_date is mishandled [airflow]
shai-ikko opened a new issue, #40841:
URL: https://github.com/apache/airflow/issues/40841
### Apache Airflow version
main (development)
### If "Other Airflow 2 version" selected, which one?
_No response_
### What happened?
Airflow has two `start_date` fields on separate structures - the Task, and the Task Instance.
The code that interprets the `wait_for_downstream` flag is part of `prev_dagrun_dep.py`; before it checks for the presence of unfinished downstream tasks, though, it looks at the task `start_date`, comparing it to the execution date of the last DAG run -- which seems perfectly logical.
https://github.com/apache/airflow/blob/63662044583031fc27d98af02f2913d324245db0/airflow/ti_deps/deps/prev_dagrun_dep.py#L159-L163
However, I'm seeing the task start_date on a sensor updated with each new DAG run. As a result, it is never less than the execution date of the last DAG run -- Airflow always thinks all instances are "the first instance of its task".
I encountered this using `S3KeySensor` for a sensor, but I've checked its code, it doesn't touch the `start_date`. Nor does any other task in my DAG, AFAICT. I've been able to reproduce the issue in a DAG I can share, that doesn't require access to a S3 service.
The problem is also reflected through the web UI -- when I open the details of a task instance after some future runs have occurred, and click "more details", I see an instance start-date that is older than the task start-date; because the task start-date is updated later.
I included more detailed investigations in the discussion, https://github.com/apache/airflow/discussions/40451
### What you think should happen instead?
As a result of the wrong task start_date, `wait_for_downstream` doesn't work.
I have a dag where the sensor is marked `wait_for_downstream=True`, looking for a file with a given suffix to show up. The task immediately downstream from the sensor renames such files, giving them a different suffix. Because of the failure, the next instance of the sensor often finds the file before it was renamed, but of course, only one of the renaming task instances can succeed; the other one fails, which triggers all sorts of handling.
### How to reproduce
Take the DAG from https://gist.github.com/shai-ikko/45fc6ae32556fbed519a0b2a3007d8a2 -- it has some more downstream processing after the problem is triggered, but I think it helps to clarify the issue.
Using Breeze, put this DAG in `files/dags/`, and create a directory `files/tmp/`. Start the DAG. To trigger runs, create files with suffix `.incoming` in the tmp directory; e.g, from out of the docker,
```console
$ touch files/tmp/use-zx81.incoming
```
Now look at the DAG on the web interface. Following the touch, I see the sensor (`detect_incoming`) succeed twice, sometimes even three times, but the next task (`send_to_processing`) can only succeed once.
Also, click one of the earlier TI runs, and check its details:
then click "more details", and compare the `start_date` in the Task Instance Attributes, to the `start_date` in the Task Attributes; you'll find that the former is earlier than the latter, and you may find the latter matching the start date of the last DAG run at the time you're looking at it.
### Operating System
Ubuntu 20.04.6 LTS
### Versions of Apache Airflow Providers
No relevant providers
### Deployment
Other Docker-based deployment
### Deployment details
```console
$ breeze down
Good version of Docker: 27.0.3.
Good version of docker-compose: 2.28.1
```
### Anything else?
This happens to me every time.
It may be relevant that my native system time is UTC+3 -- in some of the logs I've seen times reported to match UTC, and in some logs the times were 6 hours off, as it seems the native time in the dockers is UTC and something is trying to make up for that.
### Are you willing to submit PR?
- [X] 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)
--
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: commits-unsubscribe@airflow.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Task with "wait_for_downstream" doesn't wait, because start_date is mishandled [airflow]
Posted by "shai-ikko (via GitHub)" <gi...@apache.org>.
shai-ikko closed issue #40841: Task with "wait_for_downstream" doesn't wait, because start_date is mishandled
URL: https://github.com/apache/airflow/issues/40841
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Task with "wait_for_downstream" doesn't wait, because start_date is mishandled [airflow]
Posted by "omkar-foss (via GitHub)" <gi...@apache.org>.
omkar-foss commented on issue #40841:
URL: https://github.com/apache/airflow/issues/40841#issuecomment-2245709461
Hey @shai-ikko, based on the steps you provided above, I was able to reproduce this issue in my local Airflow.
I tried out your sensor DAG and observed that the dep check function [here](https://github.com/apache/airflow/blob/main/airflow/ti_deps/deps/prev_dagrun_dep.py#L159-L163) returns with a passing status for task instances in simultaneously running dagruns and never reaches the `wait_for_downstream` check [here](https://github.com/apache/airflow/blob/main/airflow/ti_deps/deps/prev_dagrun_dep.py#L190-L199), causing a race condition between the task instances of the dagruns. Eventually one dagrun succeeds while the other fails.
I'm attempting to fix this issue in [this PR](https://github.com/apache/airflow/pull/40963) by skipping [this check](https://github.com/apache/airflow/blob/main/airflow/ti_deps/deps/prev_dagrun_dep.py#L159-L163) if the last dagrun is still running, so the `wait_for_downstream` check is reached and failing status is returned where required.
I've tested this PR alongwith your sensor DAG and 10+ simultaneous file touches, and it works as expected. You can also try it out, hope this helps.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Task with "wait_for_downstream" doesn't wait, because start_date is mishandled [airflow]
Posted by "shai-ikko (via GitHub)" <gi...@apache.org>.
shai-ikko commented on issue #40841:
URL: https://github.com/apache/airflow/issues/40841#issuecomment-2247433686
Hi @omkar-foss , thanks for this!
Indeed, I've checked again, and the `start_date` on the DAG itself is updated -- so it does not represent "this is when this DAG came into being", but rather "start_date of last DAG run". And then, it being updated on the task to match the last task instance execution also makes sense.
But then, I suspect the code in `prev_dagrun_dep.py` needs a different fix -- it should drop the assumption that `start_date` means "a date before which the task did not exist"; the problem with the test you fixed is not a race condition, but that it's checking the wrong thing.
On a first guess, I'd think that the right thing to check is, as in the next check in the file --
```python
not self._has_tis(last_dagrun, ti.task_id, session=session)
```
but looking at that block makes me think that either I'm missing something, or that block is actually wrong too.
https://github.com/omkar-foss/airflow/blob/2575b39e7876b19c1be93626e2247926327a1205/airflow/ti_deps/deps/prev_dagrun_dep.py#L169-L184
(but if the block is right, then I think the "check for first instance of task when there was a run of the dag" block should be folded into it)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Task with "wait_for_downstream" doesn't wait, because start_date is mishandled [airflow]
Posted by "shai-ikko (via GitHub)" <gi...@apache.org>.
shai-ikko commented on issue #40841:
URL: https://github.com/apache/airflow/issues/40841#issuecomment-2646500221
Yep. This was mostly a user error. Sorry for the noise.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Task with "wait_for_downstream" doesn't wait, because start_date is mishandled [airflow]
Posted by "vikramkoka (via GitHub)" <gi...@apache.org>.
vikramkoka commented on issue #40841:
URL: https://github.com/apache/airflow/issues/40841#issuecomment-2641286925
@shai-ikko It seems that this issue can be closed now?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Task with "wait_for_downstream" doesn't wait, because start_date is mishandled [airflow]
Posted by "shai-ikko (via GitHub)" <gi...@apache.org>.
shai-ikko commented on issue #40841:
URL: https://github.com/apache/airflow/issues/40841#issuecomment-2236365767
There may have been some misunderstanding here - I am willing, in principle, to submit a PR, but I have no clue about the fix. The bug submission form promised that guidance may be available.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Task with "wait_for_downstream" doesn't wait, because start_date is mishandled [airflow]
Posted by "boring-cyborg[bot] (via GitHub)" <gi...@apache.org>.
boring-cyborg[bot] commented on issue #40841:
URL: https://github.com/apache/airflow/issues/40841#issuecomment-2233573341
Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Task with "wait_for_downstream" doesn't wait, because start_date is mishandled [airflow]
Posted by "omkar-foss (via GitHub)" <gi...@apache.org>.
omkar-foss commented on issue #40841:
URL: https://github.com/apache/airflow/issues/40841#issuecomment-2255245212
Hey @shai-ikko, good to hear from you. Apologies on my delayed response here.
So in [this PR](https://github.com/apache/airflow/pull/40963), the check I've added - to check if last dagrun is still running - helps to resolve the issue where multiple dagruns are spawned and more than one of them try to process the same file. The issue occurs because the each task instances of each dagrun call the `_get_dep_statuses()` which returns a passing status and the task instances in than one dagruns try to process the same file (and only one of them succeeds, others fail).
I've intentionally not touched other checks below it, as those seem to be added to handle specific cases. For example, the check you mentioned - `if not self._has_tis(last_dagrun, ti.task_id, session=session)` - **I'd tried moving this check above the `last_dagrun.execution_date < ti.task.start_date` check without any other changes, and it results in blocking new dagruns from getting scheduled after the current ones are complete** (see screenshot below).
Tagging two of the pros I'm aware of, who might be able to help here - @ashb @uranusjr please help with these couple of things, if possible:
1. Kindly help verify if [this PR](https://github.com/apache/airflow/pull/40963) will suffice to resolve this GitHub issue, or if some other changes would be required.
2. Kindly share if you happen to have any context about the reasoning behind updating `task.start_date`, as [this check](https://github.com/apache/airflow/blob/main/airflow/ti_deps/deps/prev_dagrun_dep.py#L159-L163) returns `True` for more than one task instances in parallel DagRuns, causing them to process the same file.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Task with "wait_for_downstream" doesn't wait, because start_date is mishandled [airflow]
Posted by "shai-ikko (via GitHub)" <gi...@apache.org>.
shai-ikko commented on issue #40841:
URL: https://github.com/apache/airflow/issues/40841#issuecomment-2256026832
Hi, I've looked a little more into documentation and code.
Since I think the `start_date`/`execution_date` comparison is utterly wrong, I've tried just commenting it out. I got, basically, correct behavior.
I found out that:
- [`execution_date` is probably not what you think it is](https://airflow.apache.org/docs/apache-airflow/stable/faq.html#what-does-execution-date-mean)
- `datetime.today()`, which I used for the DAG `start_date`, isn't what I thought either -- it is equivalent to `datetime.now()`, except it's naive, and since it's re-evaluated every time the DAG is parsed -- and it's parsed all the time -- my use of it in the DAG explains a lot of the weirdness.
In fact, if I replace that `datetime.today()` with "today at midnight", with existing code, I already get correct behavior.
(I still think that check is wrong, but the "start_date is mishandled" is all on me)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org