airflow/jobs.py
def heartbeat_callback(self, session=None):
        """Self destruct task if state has been moved away from running externally"""

        if self.terminating:
            # ensure termination if processes are created later
            self.task_runner.terminate()
            return

        self.task_instance.refresh_from_db()
        ti = self.task_instance

        fqdn = get_hostname()
        same_hostname = fqdn == ti.hostname
        same_process = ti.pid == os.getpid()

        if ti.state == State.RUNNING:
            if not same_hostname:
                self.log.warning("The recorded hostname %s "
                                 "does not match this instance's hostname "
                                 "%s", ti.hostname, fqdn)
                raise AirflowException("Hostname of job runner does not match")
            elif not same_process:
                current_pid = os.getpid()
                self.log.warning("Recorded pid %s does not match "
                                 "the current pid %s", ti.pid, current_pid)
                raise AirflowException("PID of job runner does not match")
        elif (
                self.task_runner.return_code() is None and
                hasattr(self.task_runner, 'process')
        ):
            self.log.warning(
                "State of this instance has been externally set to %s. "
                "Taking the poison pill.",
                ti.state
            )
            self.task_runner.terminate()
            self.terminating = True
def heartbeat_callback(self, session=None):
        """Self destruct task if state has been moved away from running externally"""

        if self.terminating:
            # ensure termination if processes are created later
            self.task_runner.terminate()
            return

        self.task_instance.refresh_from_db()
        ti = self.task_instance

        fqdn = get_hostname()
        same_hostname = fqdn == ti.hostname
        same_process = ti.pid == os.getpid()

        if ti.state == State.RUNNING:
            if not same_hostname:
                self.log.warning("The recorded hostname %s "
                                 "does not match this instance's hostname "
                                 "%s", ti.hostname, fqdn)
                raise AirflowException("Hostname of job runner does not match")
            elif not same_process:
                current_pid = os.getpid()
                self.log.warning("Recorded pid %s does not match "
                                 "the current pid %s", ti.pid, current_pid)
                raise AirflowException("PID of job runner does not match")
        elif (
                self.task_runner.return_code() is None and
                hasattr(self.task_runner, 'process')
        ):
            self.log.warning(
                "State of this instance has been externally set to %s. "
                "Taking the poison pill.",
                ti.state
            )
            self.task_runner.terminate()
            self.terminating = True
def heartbeat_callback ( self , session = None ) : if self . terminating : # ensure termination if processes are created later self . task_runner . terminate ( ) return self . task_instance . refresh_from_db ( ) ti = self . task_instance fqdn = get_hostname ( ) same_hostname = fqdn == ti . hostname same_process = ti . pid == os . getpid ( ) if ti . state == State . RUNNING : if not same_hostname : self . log . warning ( "The recorded hostname %s " "does not match this instance's hostname " "%s" , ti . hostname , fqdn ) raise AirflowException ( "Hostname of job runner does not match" ) elif not same_process : current_pid = os . getpid ( ) self . log . warning ( "Recorded pid %s does not match " "the current pid %s" , ti . pid , current_pid ) raise AirflowException ( "PID of job runner does not match" ) elif ( self . task_runner . return_code ( ) is None and hasattr ( self . task_runner , 'process' ) ) : self . log . warning ( "State of this instance has been externally set to %s. " "Taking the poison pill." , ti . state ) self . task_runner . terminate ( ) self . terminating = True
Self destruct task if state has been moved away from running externally
Self destruct task if state has been moved away from running externally