'Self destruct task if state has been moved away from running externally'

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

    if self.terminating:
        return
    self.task_instance.refresh_from_db()
    ti = self.task_instance
    if (ti.state == State.RUNNING):
        self.was_running = True
        fqdn = socket.getfqdn()
        if (fqdn != ti.hostname):
            logging.warning(u"The    recorded    hostname    {ti.hostname}    does    not    match    this    instance's    hostname    {fqdn}".format(**locals()))
            raise AirflowException(u'Hostname    of    job    runner    does    not    match')
        elif (not self._is_descendant_process(ti.pid)):
            current_pid = os.getpid()
            logging.warning(u'Recorded    pid    {ti.pid}    is    not    a    descendant    of    the    current    pid    {current_pid}'.format(**locals()))
            raise AirflowException(u'PID    of    job    runner    does    not    match')
    elif (self.was_running and (self.task_runner.return_code() is None) and hasattr(self.task_runner, u'process')):
        logging.warning(u'State    of    this    instance    has    been    externally    set    to    {}.    Taking    the    poison    pill.    So    long.'.format(ti.state))
        self.task_runner.terminate()
        self.terminating = True
