Introduce Heartbeat Parameter to Allow Per-LocalTaskJob Configuration (…

…#32313)

* Refactor Job Heartbeat Parameter to Allow Per-Job Configuration

This pull request introduces changes to allow users to set heartbeat expectations separately for LocalTaskJob,. his resolves a current limitation where all job types share a single configuration parameter for expected heartbeat time.


related to: #30908

airflow/config_templates/config.yml

@@ -2302,6 +2302,15 @@ scheduler:
       type: integer
       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. If this value is set to 0, the heartbeat interval will default
+        to the value of scheduler_zombie_task_threshold.
+      version_added: 2.7.0
+      type: integer
+      example: ~
+      default: "0"
     num_runs:
       description: |
         The number of times to try to schedule each DAG file

airflow/config_templates/default_airflow.cfg

@@ -1195,6 +1195,11 @@ job_heartbeat_sec = 5
 # how often the scheduler should run (in seconds).
 scheduler_heartbeat_sec = 5
 
+# The frequency (in seconds) at which the LocalTaskJob should send heartbeat signals to the
+# scheduler to notify it's still alive. If this value is set to 0, the heartbeat interval will default
+# to the value of scheduler_zombie_task_threshold.
+local_task_job_heartbeat_sec = 0
+
 # 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

@@ -157,8 +157,11 @@ def sigusr2_debug_handler(signum, frame):
         return_code = None
         try:
             self.task_runner.start()
-
-            heartbeat_time_limit = conf.getint("scheduler", "scheduler_zombie_task_threshold")
+            local_task_job_heartbeat_sec = conf.getint("scheduler", "local_task_job_heartbeat_sec")
+            if local_task_job_heartbeat_sec < 1:
+                heartbeat_time_limit = conf.getint("scheduler", "scheduler_zombie_task_threshold")
+            else:
+                heartbeat_time_limit = 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