You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by je...@apache.org on 2022/09/23 17:25:49 UTC
[airflow] branch v2-4-test updated (329f706d9e -> 53f0ffd767)
This is an automated email from the ASF dual-hosted git repository.
jedcunningham pushed a change to branch v2-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
from 329f706d9e Add unit test for log retrieval url (#26603)
new 5ae47cd45f Suppress SQLALCHEMY_TRACK_MODIFICATIONS warning in db init (#26617)
new 53f0ffd767 Fix Deferrable stuck as "scheduled" during backfill (#26205)
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
airflow/jobs/backfill_job.py | 6 ++++++
airflow/utils/db.py | 1 +
tests/jobs/test_backfill_job.py | 24 ++++++++++++++++++++++++
3 files changed, 31 insertions(+)
[airflow] 02/02: Fix Deferrable stuck as "scheduled" during backfill (#26205)
Posted by je...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jedcunningham pushed a commit to branch v2-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 53f0ffd7676c0c0a8c351568b534c4da93d6e817
Author: Ephraim Anierobi <sp...@gmail.com>
AuthorDate: Fri Sep 23 06:08:28 2022 +0100
Fix Deferrable stuck as "scheduled" during backfill (#26205)
Co-authored-by: Tzu-ping Chung <ur...@gmail.com>
(cherry picked from commit 3396d1f822caac7cbeb14e1e67679b8378a84a6c)
---
airflow/jobs/backfill_job.py | 6 ++++++
tests/jobs/test_backfill_job.py | 24 ++++++++++++++++++++++++
2 files changed, 30 insertions(+)
diff --git a/airflow/jobs/backfill_job.py b/airflow/jobs/backfill_job.py
index 4a2d3e7bd4..62bc8c2bb3 100644
--- a/airflow/jobs/backfill_job.py
+++ b/airflow/jobs/backfill_job.py
@@ -217,6 +217,12 @@ class BackfillJob(BaseJob):
tis_to_be_scheduled.append(ti)
ti_status.running.pop(reduced_key)
ti_status.to_run[ti.key] = ti
+ # special case: Deferrable task can go from DEFERRED to SCHEDULED;
+ # when that happens, we need to put it back as in UP_FOR_RESCHEDULE
+ elif ti.state == TaskInstanceState.SCHEDULED:
+ self.log.debug("Task instance %s is resumed from deferred state", ti)
+ ti_status.running.pop(ti.key)
+ ti_status.to_run[ti.key] = ti
# Batch schedule of task instances
if tis_to_be_scheduled:
diff --git a/tests/jobs/test_backfill_job.py b/tests/jobs/test_backfill_job.py
index 31ea238f5e..18fa630835 100644
--- a/tests/jobs/test_backfill_job.py
+++ b/tests/jobs/test_backfill_job.py
@@ -1415,6 +1415,30 @@ class TestBackfillJob:
ti_status.to_run.clear()
+ # test for scheduled
+ ti.set_state(State.SCHEDULED)
+ # Deferred tasks are put into scheduled by the triggerer
+ # Check that they are put into to_run
+ ti_status.running[ti.key] = ti
+ job._update_counters(ti_status=ti_status, session=session)
+ assert len(ti_status.running) == 0
+ assert len(ti_status.succeeded) == 0
+ assert len(ti_status.skipped) == 0
+ assert len(ti_status.failed) == 0
+ assert len(ti_status.to_run) == 1
+
+ ti_status.to_run.clear()
+ # test for deferred
+ # if a task is deferred and it's not yet time for the triggerer
+ # to reschedule it, we should leave it in ti_status.running
+ ti.set_state(State.DEFERRED)
+ ti_status.running[ti.key] = ti
+ job._update_counters(ti_status=ti_status, session=session)
+ assert len(ti_status.running) == 1
+ assert len(ti_status.succeeded) == 0
+ assert len(ti_status.skipped) == 0
+ assert len(ti_status.failed) == 0
+ assert len(ti_status.to_run) == 0
session.close()
def test_dag_dagrun_infos_between(self, dag_maker):
[airflow] 01/02: Suppress SQLALCHEMY_TRACK_MODIFICATIONS warning in db init (#26617)
Posted by je...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jedcunningham pushed a commit to branch v2-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 5ae47cd45fac9ff286b1e147f538fba1beccedee
Author: Daniel Standish <15...@users.noreply.github.com>
AuthorDate: Fri Sep 23 00:18:52 2022 -0700
Suppress SQLALCHEMY_TRACK_MODIFICATIONS warning in db init (#26617)
We already have it set false in the main flask app config, but now that we do db init from orm, and we create a distinct flask app for that, and it doesn't get the same setting.
Closes #26544
(cherry picked from commit 051ba159e54b992ca0111107df86b8abfd8b7279)
---
airflow/utils/db.py | 1 +
1 file changed, 1 insertion(+)
diff --git a/airflow/utils/db.py b/airflow/utils/db.py
index 843e6521b3..38887626de 100644
--- a/airflow/utils/db.py
+++ b/airflow/utils/db.py
@@ -656,6 +656,7 @@ def _create_db_from_orm(session):
def _create_flask_session_tbl():
flask_app = Flask(__name__)
flask_app.config['SQLALCHEMY_DATABASE_URI'] = conf.get('database', 'SQL_ALCHEMY_CONN')
+ flask_app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db = SQLAlchemy(flask_app)
AirflowDatabaseSessionInterface(app=flask_app, db=db, table='session', key_prefix='')
db.create_all()