LipuFei commented on issue #56050:
URL: https://github.com/apache/airflow/issues/56050#issuecomment-6218182842
I ran into the same problem. I'm running Airflow 3.3.2.
It seems like, if there are many events and the DAG is running, some new
events are deferred by the scheduler, so no new DAG run will be queued, but
then these events were kinda forgotten... Until another event for that asset
arrives and "unblocks" or "awakens" the scheduler.
Initially I thought it's related to `max_active_runs = 1`, but I also did a
PoC locally without `max_active_runs = 1` and I can still produce the problem.
I think it's also a problem even when `max_active_runs` is not set.
My PoC is very simple with two DAGs, one emitting multiple events and one
job:
```python
@dag
def poc_bug_1_send():
@task(outlets=[Asset("s3://test/")])
def task1(count: int, **context):
from airflow.sdk import Asset, Metadata
yield Metadata(
Asset("s3://my-test/"),
{"count": count}
)
task1.expand(count=[1, 2, 3, 4, 5])
@dag(schedule=[Asset("s3://test/")]) # with/without max_active_runs = 1
def poc_bug_1_job():
@task
def task1(triggering_asset_events, **context):
import time
print("Start")
time.sleep(30)
print("End")
task1()
```
**How to reproduce:**
1. Manually trigger the `send` DAG 2 to 3 times. Wait a little bit between
each trigger.
2. Wait until all `job` DAG runs to complete.
3. Navigate to `Assets` on the GUI. A few events do not have DAG runs (until
you send more events to the asset to unblock them).
<img width="1015" height="703" alt="Image"
src="https://github.com/user-attachments/assets/7e5d0405-80ed-4494-9681-30bd3a5c4b97";
/>
--
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]
