jedcunningham opened a new issue, #23435:
URL: https://github.com/apache/airflow/issues/23435
### Apache Airflow version
2.3.0 (latest released)
### What happened
I've found a DAG that will crash the scheduler:
```
@task
def hello():
return "hello"
hello.expand()
```
```
[2022-05-03 03:41:23,779] {scheduler_job.py:753} ERROR - Exception when
executing SchedulerJob._run_scheduler_loop
Traceback (most recent call last):
File
"/home/airflow/.local/lib/python3.7/site-packages/airflow/jobs/scheduler_job.py",
line 736, in _execute
self._run_scheduler_loop()
File
"/home/airflow/.local/lib/python3.7/site-packages/airflow/jobs/scheduler_job.py",
line 824, in _run_scheduler_loop
num_queued_tis = self._do_scheduling(session)
File
"/home/airflow/.local/lib/python3.7/site-packages/airflow/jobs/scheduler_job.py",
line 906, in _do_scheduling
callback_to_run = self._schedule_dag_run(dag_run, session)
File
"/home/airflow/.local/lib/python3.7/site-packages/airflow/jobs/scheduler_job.py",
line 1148, in _schedule_dag_run
schedulable_tis, callback_to_run = dag_run.update_state(session=session,
execute_callbacks=False)
File
"/home/airflow/.local/lib/python3.7/site-packages/airflow/utils/session.py",
line 68, in wrapper
return func(*args, **kwargs)
File
"/home/airflow/.local/lib/python3.7/site-packages/airflow/models/dagrun.py",
line 522, in update_state
info = self.task_instance_scheduling_decisions(session)
File
"/home/airflow/.local/lib/python3.7/site-packages/airflow/utils/session.py",
line 68, in wrapper
return func(*args, **kwargs)
File
"/home/airflow/.local/lib/python3.7/site-packages/airflow/models/dagrun.py",
line 661, in task_instance_scheduling_decisions
session=session,
File
"/home/airflow/.local/lib/python3.7/site-packages/airflow/models/dagrun.py",
line 714, in _get_ready_tis
expanded_tis, _ = schedulable.task.expand_mapped_task(self.run_id,
session=session)
File
"/home/airflow/.local/lib/python3.7/site-packages/airflow/models/mappedoperator.py",
line 609, in expand_mapped_task
operator.mul, self._resolve_map_lengths(run_id,
session=session).values()
TypeError: reduce() of empty sequence with no initial value
```
### What you think should happen instead
A user DAG shouldn't crash the scheduler. This specific case could likely be
an ImportError at parse time, but it makes me think we might be missing some
exception handling?
### How to reproduce
_No response_
### Operating System
Debian
### Versions of Apache Airflow Providers
_No response_
### Deployment
Official Apache Airflow Helm Chart
### Deployment details
_No response_
### Anything else
_No response_
### 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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
