0.2.3.7

airflow/__init__.py

@@ -1,3 +1,3 @@
 from models import DAG
 
-__version__ = "0.2.3.6"
+__version__ = "0.2.3.7"

airflow/jobs.py

@@ -217,7 +217,6 @@ class MasterJob(BaseJob):
                 for task in dag.tasks:
                     if task.adhoc:
                         continue
-                    cmd = ti.command(local=True)
                     if task.task_id not in ti_dict:
                         # Brand new task, let's get started
                         ti = TI(task, task.start_date)
@@ -225,6 +224,7 @@ class MasterJob(BaseJob):
                         if ti.is_runnable():
                             logging.debug(
                                 'First run for {ti}'.format(**locals()))
+                            cmd = ti.command(local=True)
                             executor.queue_command(ti.key, cmd)
                     else:
                         ti = ti_dict[task.task_id]
@@ -236,6 +236,7 @@ class MasterJob(BaseJob):
                             # the retry delay is met
                             if ti.is_runnable():
                                 logging.debug('Queuing retry: ' + str(ti))
+                                cmd = ti.command(local=True)
                                 executor.queue_command(ti.key, cmd)
                         else:
                             # Trying to run the next schedule
@@ -247,6 +248,7 @@ class MasterJob(BaseJob):
                             ti.refresh_from_db()
                             if ti.is_runnable():
                                 logging.debug('Queuing next run: ' + str(ti))
+                                cmd = ti.command(local=True)
                                 executor.queue_command(ti.key, cmd)
                     executor.heartbeat()
                 session.close()

setup.py

@@ -1,7 +1,7 @@
 from setuptools import setup, find_packages
 
 # Kept manually in sync with airflow.__version__
-version = '0.2.3.6'
+version = '0.2.3.7'
 
 setup(
     name='airflow',