You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by bo...@apache.org on 2017/03/09 16:32:52 UTC
incubator-airflow git commit: [AIRFLOW-931] Do not set QUEUED in
TaskInstances
Repository: incubator-airflow
Updated Branches:
refs/heads/master abbb4ee5c -> e42398100
[AIRFLOW-931] Do not set QUEUED in TaskInstances
The contract of TaskInstances stipulates that end
states for Tasks
can only be UP_FOR_RETRY, SUCCESS, FAILED,
UPSTREAM_FAILED or
SKIPPED. If concurrency was reached task instances
were set to
QUEUED by the task instance themselves. This would
prevent the
scheduler to pick them up again.
We set the state to NONE now, to ensure integrity.
Closes #2127 from bolkedebruin/AIRFLOW-931
Project: http://git-wip-us.apache.org/repos/asf/incubator-airflow/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-airflow/commit/e4239810
Tree: http://git-wip-us.apache.org/repos/asf/incubator-airflow/tree/e4239810
Diff: http://git-wip-us.apache.org/repos/asf/incubator-airflow/diff/e4239810
Branch: refs/heads/master
Commit: e42398100a3248eddb6b511ade73f6a239e58090
Parents: abbb4ee
Author: Bolke de Bruin <bo...@xs4all.nl>
Authored: Thu Mar 9 08:32:46 2017 -0800
Committer: Bolke de Bruin <bo...@xs4all.nl>
Committed: Thu Mar 9 08:32:46 2017 -0800
----------------------------------------------------------------------
airflow/models.py | 27 ++++++++++++++-------------
tests/models.py | 13 +++++++++++++
2 files changed, 27 insertions(+), 13 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/e4239810/airflow/models.py
----------------------------------------------------------------------
diff --git a/airflow/models.py b/airflow/models.py
index 37f8823..b9fda9f 100755
--- a/airflow/models.py
+++ b/airflow/models.py
@@ -1288,19 +1288,20 @@ class TaskInstance(Base):
verbose=True)
if not runnable and not mark_success:
- if self.state != State.QUEUED:
- # If a task's dependencies are met but it can't be run yet then queue it
- # instead
- self.state = State.QUEUED
- msg = "Queuing attempt {attempt} of {total}".format(
- attempt=self.try_number % (task.retries + 1) + 1,
- total=task.retries + 1)
- logging.info(hr + msg + hr)
-
- self.queued_dttm = datetime.now()
- msg = "Queuing into pool {}".format(self.pool)
- logging.info(msg)
- session.merge(self)
+ # FIXME: we might have hit concurrency limits, which means we probably
+ # have been running prematurely. This should be handled in the
+ # scheduling mechanism.
+ self.state = State.NONE
+ msg = ("FIXME: Rescheduling due to concurrency limits reached at task "
+ "runtime. Attempt {attempt} of {total}. State set to NONE.").format(
+ attempt=self.try_number % (task.retries + 1) + 1,
+ total=task.retries + 1)
+ logging.warning(hr + msg + hr)
+
+ self.queued_dttm = datetime.now()
+ msg = "Queuing into pool {}".format(self.pool)
+ logging.info(msg)
+ session.merge(self)
session.commit()
return
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/e4239810/tests/models.py
----------------------------------------------------------------------
diff --git a/tests/models.py b/tests/models.py
index d758972..9da6f16 100644
--- a/tests/models.py
+++ b/tests/models.py
@@ -366,6 +366,19 @@ class TaskInstanceTest(unittest.TestCase):
dag >> op5
self.assertIs(op5.dag, dag)
+ @patch.object(DAG, 'concurrency_reached')
+ def test_requeue_over_concurrency(self, mock_concurrency_reached):
+ mock_concurrency_reached.return_value = True
+
+ dag = DAG(dag_id='test_requeue_over_concurrency', start_date=DEFAULT_DATE,
+ max_active_runs=1, concurrency=2)
+ task = DummyOperator(task_id='test_requeue_over_concurrency_op', dag=dag)
+
+ ti = TI(task=task, execution_date=datetime.datetime.now())
+ ti.run()
+ self.assertEqual(ti.state, models.State.NONE)
+
+
@patch.object(TI, 'pool_full')
def test_run_pooling_task(self, mock_pool_full):
"""