You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "kubmichael (via GitHub)" <gi...@apache.org> on 2025/06/10 13:42:44 UTC
[I] Failure during enqueuing DAGs with dynamic tasks [airflow]
kubmichael opened a new issue, #51575:
URL: https://github.com/apache/airflow/issues/51575
### Apache Airflow version
3.0.1
### If "Other Airflow 2 version" selected, which one?
_No response_
### What happened?
Hey, we encountered failures during scheduling DAGs that create tasks dynamically. The number of tasks is 16 or more. During scheduling we observed the following exception that is causing the airflow-scheduler to restart
Airflow is deployed on Kubernetes and we are using CeleryExecutor
```
2025-06-10T13:10:38.985871927Z stdout F [2025-06-10T13:10:38.984+0000] {scheduler_job_runner.py:1019} ERROR - Exception when executing SchedulerJob._run_scheduler
_loop
2025-06-10T13:10:38.98590054Z stdout F Traceback (most recent call last):
2025-06-10T13:10:38.985903837Z stdout F File "/home/airflow/.local/lib/python3.12/site-packages/airflow/jobs/scheduler_job_runner.py", line 1015, in _execute
2025-06-10T13:10:38.985907032Z stdout F self._run_scheduler_loop()
2025-06-10T13:10:38.985910709Z stdout F File "/home/airflow/.local/lib/python3.12/site-packages/airflow/jobs/scheduler_job_runner.py", line 1304, in _run_schedu
ler_loop
2025-06-10T13:10:38.985914169Z stdout F executor.heartbeat()
2025-06-10T13:10:38.985916863Z stdout F File "/home/airflow/.local/lib/python3.12/site-packages/airflow/traces/tracer.py", line 57, in wrapper
2025-06-10T13:10:38.985919747Z stdout F return func(*args, **kwargs)
2025-06-10T13:10:38.985922292Z stdout F ^^^^^^^^^^^^^^^^^^^^^
2025-06-10T13:10:38.985925147Z stdout F File "/home/airflow/.local/lib/python3.12/site-packages/airflow/executors/base_executor.py", line 282, in heartbeat
2025-06-10T13:10:38.985927878Z stdout F self.trigger_tasks(open_slots)
2025-06-10T13:10:38.985930829Z stdout F File "/home/airflow/.local/lib/python3.12/site-packages/airflow/traces/tracer.py", line 57, in wrapper
2025-06-10T13:10:38.985933536Z stdout F return func(*args, **kwargs)
2025-06-10T13:10:38.985936032Z stdout F ^^^^^^^^^^^^^^^^^^^^^
2025-06-10T13:10:38.985938523Z stdout F File "/home/airflow/.local/lib/python3.12/site-packages/airflow/executors/base_executor.py", line 446, in trigger_tasks
2025-06-10T13:10:38.98594104Z stdout F span = Trace.start_child_span(
2025-06-10T13:10:38.985943645Z stdout F ^^^^^^^^^^^^^^^^^^^^^^^
2025-06-10T13:10:38.985946239Z stdout F File "/home/airflow/.local/lib/python3.12/site-packages/airflow/traces/otel_tracer.py", line 215, in start_child_span
2025-06-10T13:10:38.985949153Z stdout F context_val = next(iter(parent_context.values()))
2025-06-10T13:10:38.985951717Z stdout F ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
2025-06-10T13:10:38.985964247Z stdout F StopIteration
2025-06-10T13:10:38.986460605Z stdout F [2025-06-10T13:10:38.986+0000] {scheduler_job_runner.py:1031} INFO - Exited execute loop
2025-06-10T13:10:39.004345675Z stderr F Traceback (most recent call last):
2025-06-10T13:10:39.004369761Z stderr F File "/home/airflow/.local/bin/airflow", line 8, in <module>
2025-06-10T13:10:39.004429823Z stderr F sys.exit(main())
2025-06-10T13:10:39.004466357Z stderr F ^^^^^^
2025-06-10T13:10:39.00446883Z stderr F File "/home/airflow/.local/lib/python3.12/site-packages/airflow/__main__.py", line 55, in main
2025-06-10T13:10:39.004672407Z stderr F args.func(args)
2025-06-10T13:10:39.004685243Z stderr F File "/home/airflow/.local/lib/python3.12/site-packages/airflow/cli/cli_config.py", line 48, in command
2025-06-10T13:10:39.004778388Z stderr F return func(*args, **kwargs)
2025-06-10T13:10:39.00478586Z stderr F ^^^^^^^^^^^^^^^^^^^^^
2025-06-10T13:10:39.004788161Z stderr F File "/home/airflow/.local/lib/python3.12/site-packages/airflow/utils/cli.py", line 111, in wrapper
2025-06-10T13:10:39.004938927Z stderr F return f(*args, **kwargs)
2025-06-10T13:10:39.0049886Z stderr F ^^^^^^^^^^^^^^^^^^
2025-06-10T13:10:39.004993856Z stderr F File "/home/airflow/.local/lib/python3.12/site-packages/airflow/utils/providers_configuration_loader.py", line 55, in wrapped_function
2025-06-10T13:10:39.005028116Z stderr F return func(*args, **kwargs)
2025-06-10T13:10:39.005126992Z stderr F ^^^^^^^^^^^^^^^^^^^^^
2025-06-10T13:10:39.005133438Z stderr F File "/home/airflow/.local/lib/python3.12/site-packages/airflow/cli/commands/scheduler_command.py", line 52, in scheduler
2025-06-10T13:10:39.005196617Z stderr F run_command_with_daemon_option(
2025-06-10T13:10:39.005360652Z stderr F File "/home/airflow/.local/lib/python3.12/site-packages/airflow/cli/commands/daemon_utils.py", line 86, in run_command_w
ith_daemon_option
2025-06-10T13:10:39.005420529Z stderr F callback()
2025-06-10T13:10:39.005470811Z stderr F File "/home/airflow/.local/lib/python3.12/site-packages/airflow/cli/commands/scheduler_command.py", line 55, in <lambda>
2025-06-10T13:10:39.005518232Z stderr F callback=lambda: _run_scheduler_job(args),
2025-06-10T13:10:39.005568225Z stderr F ^^^^^^^^^^^^^^^^^^^^^^^^
2025-06-10T13:10:39.005572002Z stderr F File "/home/airflow/.local/lib/python3.12/site-packages/airflow/cli/commands/scheduler_command.py", line 43, in _run_sch
eduler_job
2025-06-10T13:10:39.005621626Z stderr F run_job(job=job_runner.job, execute_callable=job_runner._execute)
2025-06-10T13:10:39.005671978Z stderr F File "/home/airflow/.local/lib/python3.12/site-packages/airflow/utils/session.py", line 101, in wrapper
2025-06-10T13:10:39.005719365Z stderr F return func(*args, session=session, **kwargs)
2025-06-10T13:10:39.005797109Z stderr F ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
2025-06-10T13:10:39.005810073Z stderr F File "/home/airflow/.local/lib/python3.12/site-packages/airflow/jobs/job.py", line 347, in run_job
2025-06-10T13:10:39.005921519Z stderr F return execute_job(job, execute_callable=execute_callable)
2025-06-10T13:10:39.00592605Z stderr F ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
2025-06-10T13:10:39.00592928Z stderr F File "/home/airflow/.local/lib/python3.12/site-packages/airflow/jobs/job.py", line 376, in execute_job
2025-06-10T13:10:39.006062941Z stderr F ret = execute_callable()
2025-06-10T13:10:39.006067891Z stderr F ^^^^^^^^^^^^^^^^^^
2025-06-10T13:10:39.006071284Z stderr F File "/home/airflow/.local/lib/python3.12/site-packages/airflow/jobs/scheduler_job_runner.py", line 1015, in _execute
2025-06-10T13:10:39.006331311Z stderr F self._run_scheduler_loop()
2025-06-10T13:10:39.006338198Z stderr F File "/home/airflow/.local/lib/python3.12/site-packages/airflow/jobs/scheduler_job_runner.py", line 1304, in _run_schedu
ler_loop
2025-06-10T13:10:39.006686485Z stderr F executor.heartbeat()
2025-06-10T13:10:39.006703837Z stderr F File "/home/airflow/.local/lib/python3.12/site-packages/airflow/traces/tracer.py", line 57, in wrapper
2025-06-10T13:10:39.006864212Z stderr F return func(*args, **kwargs)
2025-06-10T13:10:39.007195779Z stderr F ^^^^^^^^^^^^^^^^^^^^^
2025-06-10T13:10:39.007204227Z stderr F File "/home/airflow/.local/lib/python3.12/site-packages/airflow/executors/base_executor.py", line 282, in heartbeat
2025-06-10T13:10:39.007207267Z stderr F self.trigger_tasks(open_slots)
2025-06-10T13:10:39.00721514Z stderr F File "/home/airflow/.local/lib/python3.12/site-packages/airflow/traces/tracer.py", line 57, in wrapper
2025-06-10T13:10:39.00728819Z stderr F return func(*args, **kwargs)
2025-06-10T13:10:39.007292538Z stderr F ^^^^^^^^^^^^^^^^^^^^^
2025-06-10T13:10:39.007295402Z stderr F File "/home/airflow/.local/lib/python3.12/site-packages/airflow/executors/base_executor.py", line 446, in trigger_tasks
2025-06-10T13:10:39.007489356Z stderr F span = Trace.start_child_span(
2025-06-10T13:10:39.007498452Z stderr F ^^^^^^^^^^^^^^^^^^^^^^^
2025-06-10T13:10:39.00750522Z stderr F File "/home/airflow/.local/lib/python3.12/site-packages/airflow/traces/otel_tracer.py", line 215, in start_child_span
2025-06-10T13:10:39.007593246Z stderr F context_val = next(iter(parent_context.values()))
2025-06-10T13:10:39.007645859Z stderr F ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
2025-06-10T13:10:39.00764971Z stderr F StopIteration
```
### What you think should happen instead?
We are expecting that DAGs will be enqueued and the scheduler won't enter a restart loop
### How to reproduce
Run two or more DAGs that create dynamic tasks,each more than 16
### Operating System
Amazon Linux 2023.7.20250331
### Versions of Apache Airflow Providers
apache-airflow-providers-amazon 9.8.0
apache-airflow-providers-celery 3.10.6
apache-airflow-providers-cncf-kubernetes 10.4.3
apache-airflow-providers-common-compat 1.6.1
apache-airflow-providers-common-io 1.5.4
apache-airflow-providers-common-messaging 1.0.1
apache-airflow-providers-common-sql 1.27.0
apache-airflow-providers-docker 4.3.1
apache-airflow-providers-elasticsearch 6.2.2
apache-airflow-providers-fab 2.0.2
apache-airflow-providers-ftp 3.12.3
apache-airflow-providers-git 0.0.2
apache-airflow-providers-google 15.1.0
apache-airflow-providers-grpc 3.7.3
apache-airflow-providers-hashicorp 4.1.1
apache-airflow-providers-http 5.2.2
apache-airflow-providers-microsoft-azure 12.3.1
apache-airflow-providers-mysql 6.2.2
apache-airflow-providers-odbc 4.9.2
apache-airflow-providers-openlineage 2.2.0
apache-airflow-providers-postgres 6.1.3
apache-airflow-providers-redis 4.0.2
apache-airflow-providers-sendgrid 4.0.1
apache-airflow-providers-sftp 5.2.1
apache-airflow-providers-slack 9.0.5
apache-airflow-providers-smtp 2.0.3
apache-airflow-providers-snowflake 6.3.0
apache-airflow-providers-ssh 4.0.1
apache-airflow-providers-standard 1.1.0
### Deployment
Other
### Deployment details
We are using kubernetes,
for helm chart we are doing following, because official chart isn't release yet
```
git clone --depth 1 https://github.com/apache/airflow.git airflow-og-repo
helm upgrade --install airflow airflow-og-repo/chart --namespace airflow
```
### 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)
--
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] Failure during enqueuing DAGs with dynamic tasks [airflow]
Posted by "kubmichael (via GitHub)" <gi...@apache.org>.
kubmichael commented on issue #51575:
URL: https://github.com/apache/airflow/issues/51575#issuecomment-3161902663
@jroachgolf84 Hey, sorry for the delay. We do see that this issue still exists. The scheduler performs multiple restarts and succeeds at some point in time.
--
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] Failure during enqueuing DAGs with dynamic tasks [airflow]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed issue #51575: Failure during enqueuing DAGs with dynamic tasks
URL: https://github.com/apache/airflow/issues/51575
--
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] Failure during enqueuing DAGs with dynamic tasks [airflow]
Posted by "kubmichael (via GitHub)" <gi...@apache.org>.
kubmichael commented on issue #51575:
URL: https://github.com/apache/airflow/issues/51575#issuecomment-2973644986
> Would you be able to share a snippet of the code that you're running to do this? Please feel free to strip out any proprietary information, I'm mostly interested in the DAG that's generating the value that's eventually passed to `.expand()`.
Hey, this is a code of dag
`@dag(
"snapshot_scraper",
schedule="0 7 * * *",
default_args={"depends_on_past": False},
start_date=START_DATE,
catchup=False,
)
def snapshot_scraper_dag():
t_start = EmptyOperator(task_id="start")
@task_group
def snapshot_scraper_tg(scraper_config: SnapshotScraperConfig):
@task
def scrape(config: SnapshotScraperConfig, **context):
prev_start_date_success = context["prev_start_date_success"]
logger.info(f"scraping snapshot {config.name}, {prev_start_date_success}")
from utils import load_secret_env
load_secret_env()
run_helper.scrape_snapshot(config, _parse_date(prev_start_date_success))
@task
def parse(config: SnapshotScraperConfig, **context):
prev_start_date_success = context["prev_start_date_success"]
logger.info(f"parsing snapshot {config.name}, {prev_start_date_success}")
from utils import load_secret_env
load_secret_env()
run_helper.parse_snapshot(config, _parse_date(prev_start_date_success))
scrape(scraper_config) >> parse(scraper_config) # type: ignore
snapshot_scraper_tg_obj = snapshot_scraper_tg.expand(
scraper_config=list(load_snapshot_configs())
)
t_end = EmptyOperator(task_id="end", trigger_rule=TriggerRule.ALL_DONE)
t_start >> snapshot_scraper_tg_obj >> t_end
snapshot_scraper_dag()
`
and
`
def load_snapshot_configs():
for domain in _load_yaml("./snapshots.yaml"):
for space in domain.get("spaces", []):
yield SnapshotScraperConfig(
domain=domain["domain"],
name=space["name"],
space_id=space["space_id"],
)
`
So, we have two dags that creates task dynamically, both have more than 16 tasks, and when they scheduled at the same time airflow-dag-processor can't enqueue 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] Failure during enqueuing DAGs with dynamic tasks [airflow]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on issue #51575:
URL: https://github.com/apache/airflow/issues/51575#issuecomment-3230316736
This issue has been closed because it has not received response from the issue author.
--
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] Failure during enqueuing DAGs with dynamic tasks [airflow]
Posted by "jroachgolf84 (via GitHub)" <gi...@apache.org>.
jroachgolf84 commented on issue #51575:
URL: https://github.com/apache/airflow/issues/51575#issuecomment-3140983631
@kubmichael, just wanted to follow up on this. Can you try with 3.0.3?
--
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] Failure during enqueuing DAGs with dynamic tasks [airflow]
Posted by "vatsrahul1001 (via GitHub)" <gi...@apache.org>.
vatsrahul1001 commented on issue #51575:
URL: https://github.com/apache/airflow/issues/51575#issuecomment-3130892487
> cc: [@vatsrahul1001](https://github.com/vatsrahul1001) I think you had a test case for this earlier?
Yes Vikram
--
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] Failure during enqueuing DAGs with dynamic tasks [airflow]
Posted by "jroachgolf84 (via GitHub)" <gi...@apache.org>.
jroachgolf84 commented on issue #51575:
URL: https://github.com/apache/airflow/issues/51575#issuecomment-2959698831
Would you be able to share a snippet of the code that you're running to do this? Please feel free to strip out any proprietary information, I'm mostly interested in the DAG that's generating the value that's eventually passed to `.expand()`.
--
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] Failure during enqueuing DAGs with dynamic tasks [airflow]
Posted by "vikramkoka (via GitHub)" <gi...@apache.org>.
vikramkoka commented on issue #51575:
URL: https://github.com/apache/airflow/issues/51575#issuecomment-3130195097
cc: @vatsrahul1001
I think you had a test case for this earlier?
--
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] Failure during enqueuing DAGs with dynamic tasks [airflow]
Posted by "vatsrahul1001 (via GitHub)" <gi...@apache.org>.
vatsrahul1001 commented on issue #51575:
URL: https://github.com/apache/airflow/issues/51575#issuecomment-3130893748
@kubmichael, I am not able to replicate this. Can you try this with 3.0.3?
--
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] Failure during enqueuing DAGs with dynamic tasks [airflow]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on issue #51575:
URL: https://github.com/apache/airflow/issues/51575#issuecomment-3208520055
This issue has been automatically marked as stale because it has been open for 14 days with no response from the author. It will be closed in next 7 days if no further activity occurs from the issue author.
--
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