Introduce Heartbeat Parameter to Allow Per-LocalTaskJob Configuration

airflow/config_templates/config.yml

@@ -2268,6 +2268,15 @@ scheduler:
       type: string
       example: ~
       default: "5"
+    local_task_job_heartbeat_sec:
+      description: |
+        The frequency (in seconds) at which the LocalTaskJob should send heartbeat signals to the
+        scheduler to notify it's still alive. The value of this parameter should be less than the 
+        scheduler's `scheduler_zombie_task_threshold` config.
+      version_added: ~
+      type: int
+      example: ~
+      default: "5"
     num_runs:
       description: |
         The number of times to try to schedule each DAG file

airflow/config_templates/default_airflow.cfg

@@ -1173,6 +1173,12 @@ job_heartbeat_sec = 5
 # how often the scheduler should run (in seconds).
 scheduler_heartbeat_sec = 5
 
+# The heartbeat frequency (in seconds) at which the scheduler will check the state of the LocalTaskJob.
+# A high value could lead to the scheduler not noticing a "zombie" LocalTaskJob quickly,
+# while a low value could mean more frequent database queries.
+# This setting is used to detect tasks that hang or tasks that have been erroneously marked as running but are not actually running.
+local_task_job_heartbeat_sec = 5
+
 # The number of times to try to schedule each DAG file
 # -1 indicates unlimited number
 num_runs = -1

airflow/jobs/local_task_job_runner.py

@@ -158,7 +158,7 @@ def sigusr2_debug_handler(signum, frame):
         try:
             self.task_runner.start()
 
-            heartbeat_time_limit = conf.getint("scheduler", "scheduler_zombie_task_threshold")
+            heartbeat_time_limit = conf.getint("scheduler", "local_task_job_heartbeat_sec")
 
             # LocalTaskJob should not run callbacks, which are handled by TaskInstance._run_raw_task
             # 1, LocalTaskJob does not parse DAG, thus cannot run callbacks