self.task_runner.terminate()
        self.task_runner.on_finish()
    @provide_session
    def heartbeat_callback(self, session: Session = NEW_SESSION) -> None:
        """Self destruct task if state has been moved away from running externally."""
        if self.terminating:
            
            self.task_runner.terminate()
            return
        self.task_instance.refresh_from_db()
        ti = self.task_instance
        if ti.state == TaskInstanceState.RUNNING:
            fqdn = get_hostname()
            same_hostname = fqdn == ti.hostname
            if not same_hostname:
                self.log.error(
                    "The recorded hostname %s does not match this instance's hostname %s",
                    ti.hostname,
                    fqdn,
                )
                raise AirflowException("Hostname of job runner does not match")
            current_pid = self.task_runner.get_process_pid()
            recorded_pid = ti.pid
            same_process = recorded_pid == current_pid
            if recorded_pid is not None and (ti.run_as_user or self.task_runner.run_as_user):