class_docstr: LocalTaskJob runs a single task instance.
class_docstr_tok: ["LocalTaskJob","runs","a","single","task","instance","."]
docstr: Self destruct task if state has been moved away from running externally.
docstr_tok: ["Self","destruct","task","if","state","has","been","moved","away","from","running","externally","."]
code: def heartbeat_callback(self, session: Session = NEW_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
  
      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
          ):
              # when running as another user, compare the task runner pid to the parent of
              # the recorded pid because user delegation becomes an extra process level.
              # However, if recorded_pid is None, pass that through as it signals the task
              # runner process has already completed and been cleared out. `psutil.Process`
              # uses the current process if the parameter is None, which is not what is intended
              # for comparison.
              recorded_pid = psutil.Process(ti.pid).ppid()
              same_process = recorded_pid == current_pid
  
          if (
              recorded_pid is not None
              and not same_process
              and not IS_WINDOWS
          ):
              self.log.warning(
                  "Recorded pid %s does not match the current pid %s",
                  recorded_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"
      ):
          if ti.state == TaskInstanceState.SKIPPED:
              # A DagRun timeout will cause tasks to be externally marked as skipped.
              dagrun = ti.get_dagrun(session=session)
              execution_time = (
                  dagrun.end_date or timezone.utcnow()
              ) - (dagrun.start_date or timezone.utcnow())
              if ti.task.dag is not None:
                  dagrun_timeout = ti.task.dag.dagrun_timeout
              else:
                  dagrun_timeout = None
              if dagrun_timeout and execution_time > dagrun_timeout:
                  self.log.warning(
                      "DagRun timed out after %s.", execution_time
                  )
  
          # potential race condition, the _run_raw_task commits `success` or other state
          # but task_runner does not exit right away due to slow process shutdown or any other reasons
          # let's do a throttle here, if the above case is true, the handle_task_exit will handle it
          if (
              self._state_change_checks >= 1
          ):  # defer to next round of heartbeat
              self.log.warning(
                  "State of this instance has been externally set to %s. Terminating instance.",
                  ti.state,
              )
              self.terminating = True
          self._state_change_checks += 1
  
imports: from __future__ import annotations
  import signal
  from typing import TYPE_CHECKING
  import psutil
  from airflow.configuration import conf
  from airflow.exceptions import AirflowException
  from airflow.jobs.base_job_runner import BaseJobRunner
  from airflow.jobs.job import perform_heartbeat
  from airflow.models.taskinstance import TaskReturnCode
  from airflow.stats import Stats
  from airflow.utils import timezone
  from airflow.utils.log.file_task_handler import _set_task_deferred_context_var
  from airflow.utils.log.logging_mixin import LoggingMixin
  from airflow.utils.net import get_hostname
  from airflow.utils.platform import IS_WINDOWS
  from airflow.utils.session import NEW_SESSION, provide_session
  from airflow.utils.state import TaskInstanceState
label_desc: Use image node_id 5 for calling the LocalTaskJobRunner obj's underlying member method code with example usage: obj.heartbeat_callback(session) without return types