Only compare updated time when Serialized DAG exists #13899
Merged
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
closes apache#13667 The following error happens when Serialized DAGs exist in Webserver or Scheduler but it has just been removed from serialized_dag table, mainly due to the removal of DAG file. ``` Traceback (most recent call last): File "/home/app/.pyenv/versions/3.8.1/envs/airflow-py381/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 1275, in _execute self._run_scheduler_loop() File "/home/app/.pyenv/versions/3.8.1/envs/airflow-py381/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 1377, in _run_scheduler_loop num_queued_tis = self._do_scheduling(session) File "/home/app/.pyenv/versions/3.8.1/envs/airflow-py381/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 1516, in _do_scheduling self._schedule_dag_run(dag_run, active_runs_by_dag_id.get(dag_run.dag_id, set()), session) File "/home/app/.pyenv/versions/3.8.1/envs/airflow-py381/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 1629, in _schedule_dag_run dag = dag_run.dag = self.dagbag.get_dag(dag_run.dag_id, session=session) File "/home/app/.pyenv/versions/3.8.1/envs/airflow-py381/lib/python3.8/site-packages/airflow/utils/session.py", line 62, in wrapper return func(*args, **kwargs) File "/home/app/.pyenv/versions/3.8.1/envs/airflow-py381/lib/python3.8/site-packages/airflow/models/dagbag.py", line 187, in get_dag if sd_last_updated_datetime > self.dags_last_fetched[dag_id] ``` A simple fix is to just check if `sd_last_updated_datetime` is not `None` i.e. Serialized DAG for that dag_id is not None
XD-DENG
approved these changes
Jan 26, 2021
kaxil
added a commit
that referenced
this pull request
Jan 27, 2021
closes #13667 The following error happens when Serialized DAGs exist in Webserver or Scheduler but it has just been removed from serialized_dag table, mainly due to the removal of DAG file. ``` Traceback (most recent call last): File "/home/app/.pyenv/versions/3.8.1/envs/airflow-py381/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 1275, in _execute self._run_scheduler_loop() File "/home/app/.pyenv/versions/3.8.1/envs/airflow-py381/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 1377, in _run_scheduler_loop num_queued_tis = self._do_scheduling(session) File "/home/app/.pyenv/versions/3.8.1/envs/airflow-py381/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 1516, in _do_scheduling self._schedule_dag_run(dag_run, active_runs_by_dag_id.get(dag_run.dag_id, set()), session) File "/home/app/.pyenv/versions/3.8.1/envs/airflow-py381/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 1629, in _schedule_dag_run dag = dag_run.dag = self.dagbag.get_dag(dag_run.dag_id, session=session) File "/home/app/.pyenv/versions/3.8.1/envs/airflow-py381/lib/python3.8/site-packages/airflow/utils/session.py", line 62, in wrapper return func(*args, **kwargs) File "/home/app/.pyenv/versions/3.8.1/envs/airflow-py381/lib/python3.8/site-packages/airflow/models/dagbag.py", line 187, in get_dag if sd_last_updated_datetime > self.dags_last_fetched[dag_id] ``` A simple fix is to just check if `sd_last_updated_datetime` is not `None` i.e. Serialized DAG for that dag_id is not None (cherry picked from commit 8958d12)
kaxil
added a commit
that referenced
this pull request
Feb 4, 2021
closes #13667 The following error happens when Serialized DAGs exist in Webserver or Scheduler but it has just been removed from serialized_dag table, mainly due to the removal of DAG file. ``` Traceback (most recent call last): File "/home/app/.pyenv/versions/3.8.1/envs/airflow-py381/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 1275, in _execute self._run_scheduler_loop() File "/home/app/.pyenv/versions/3.8.1/envs/airflow-py381/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 1377, in _run_scheduler_loop num_queued_tis = self._do_scheduling(session) File "/home/app/.pyenv/versions/3.8.1/envs/airflow-py381/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 1516, in _do_scheduling self._schedule_dag_run(dag_run, active_runs_by_dag_id.get(dag_run.dag_id, set()), session) File "/home/app/.pyenv/versions/3.8.1/envs/airflow-py381/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 1629, in _schedule_dag_run dag = dag_run.dag = self.dagbag.get_dag(dag_run.dag_id, session=session) File "/home/app/.pyenv/versions/3.8.1/envs/airflow-py381/lib/python3.8/site-packages/airflow/utils/session.py", line 62, in wrapper return func(*args, **kwargs) File "/home/app/.pyenv/versions/3.8.1/envs/airflow-py381/lib/python3.8/site-packages/airflow/models/dagbag.py", line 187, in get_dag if sd_last_updated_datetime > self.dags_last_fetched[dag_id] ``` A simple fix is to just check if `sd_last_updated_datetime` is not `None` i.e. Serialized DAG for that dag_id is not None (cherry picked from commit 8958d12)
kaxil
added a commit
to astronomer/airflow
that referenced
this pull request
Feb 16, 2021
closes apache#13667 The following error happens when Serialized DAGs exist in Webserver or Scheduler but it has just been removed from serialized_dag table, mainly due to the removal of DAG file. ``` Traceback (most recent call last): File "/home/app/.pyenv/versions/3.8.1/envs/airflow-py381/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 1275, in _execute self._run_scheduler_loop() File "/home/app/.pyenv/versions/3.8.1/envs/airflow-py381/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 1377, in _run_scheduler_loop num_queued_tis = self._do_scheduling(session) File "/home/app/.pyenv/versions/3.8.1/envs/airflow-py381/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 1516, in _do_scheduling self._schedule_dag_run(dag_run, active_runs_by_dag_id.get(dag_run.dag_id, set()), session) File "/home/app/.pyenv/versions/3.8.1/envs/airflow-py381/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 1629, in _schedule_dag_run dag = dag_run.dag = self.dagbag.get_dag(dag_run.dag_id, session=session) File "/home/app/.pyenv/versions/3.8.1/envs/airflow-py381/lib/python3.8/site-packages/airflow/utils/session.py", line 62, in wrapper return func(*args, **kwargs) File "/home/app/.pyenv/versions/3.8.1/envs/airflow-py381/lib/python3.8/site-packages/airflow/models/dagbag.py", line 187, in get_dag if sd_last_updated_datetime > self.dags_last_fetched[dag_id] ``` A simple fix is to just check if `sd_last_updated_datetime` is not `None` i.e. Serialized DAG for that dag_id is not None (cherry picked from commit 8958d12) (cherry picked from commit 3881ad4)
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
closes #13667
The following error happens when Serialized DAGs exist in Webserver or Scheduler but it has just been removed from serialized_dag table,
mainly due to the removal of DAG file.
A simple fix is to just check if
sd_last_updated_datetimeis notNonei.e. Serialized DAG for that dag_id is not None^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code change, Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in UPDATING.md.