'Fails tasks that haven\'t had a heartbeat in too long'

@provide_session
def kill_zombies(self, session=None):

    from airflow.jobs import LocalTaskJob as LJ
    self.logger.info(u"Finding    'running'    jobs    without    a    recent    heartbeat")
    TI = TaskInstance
    secs = configuration.getint(u'scheduler', u'scheduler_zombie_task_threshold')
    limit_dttm = (datetime.now() - timedelta(seconds=secs))
    self.logger.info(u'Failing    jobs    without    heartbeat    after    {}'.format(limit_dttm))
    tis = session.query(TI).join(LJ, (TI.job_id == LJ.id)).filter((TI.state == State.RUNNING)).filter(or_((LJ.state != State.RUNNING), (LJ.latest_heartbeat < limit_dttm))).all()
    for ti in tis:
        if (ti and (ti.dag_id in self.dags)):
            dag = self.dags[ti.dag_id]
            if (ti.task_id in dag.task_ids):
                task = dag.get_task(ti.task_id)
                ti.task = task
                ti.handle_failure(u'{}    killed    as    zombie'.format(ti))
                self.logger.info(u'Marked    zombie    job    {}    as    failed'.format(ti))
                Stats.incr(u'zombies_killed')
    session.commit()
