Update zombie message to be more descriptive

airflow/jobs/scheduler_job.py

@@ -1481,11 +1481,31 @@ def _find_zombies(self, session: Session) -> None:
             self.log.warning("Failing (%s) jobs without heartbeat after %s", len(zombies), limit_dttm)
 
         for ti, file_loc in zombies:
+
+            zombie_message_details = self._generate_zombie_message_details(ti)
             request = TaskCallbackRequest(
                 full_filepath=file_loc,
                 simple_task_instance=SimpleTaskInstance.from_ti(ti),
-                msg=f"Detected {ti} as zombie",
+                msg=str(zombie_message_details),
             )
-            self.log.error("Detected zombie job: %s", request)
+
+            self.log.error("Detected zombie job: %s", request.msg)
             self.executor.send_callback(request)
             Stats.incr('zombies_killed')
+
+    @staticmethod
+    def _generate_zombie_message_details(ti: TaskInstance):
+        zombie_message_details = {
+            "DAG Id": ti.dag_id,
+            "Task Id": ti.task_id,
+            "Run Id": ti.run_id,
+        }
+
+        if ti.map_index != -1:
+            zombie_message_details["Map Index"] = ti.map_index
+        if ti.hostname:
+            zombie_message_details["Hostname"] = ti.hostname
+        if ti.external_executor_id:
+            zombie_message_details["External Executor Id"] = ti.external_executor_id
+
+        return zombie_message_details

tests/jobs/test_scheduler_job.py

@@ -4044,7 +4044,7 @@ def test_find_zombies(self):
             requests = self.scheduler_job.executor.callback_sink.send.call_args[0]
             assert 1 == len(requests)
             assert requests[0].full_filepath == dag.fileloc
-            assert requests[0].msg == f"Detected {ti} as zombie"
+            assert requests[0].msg == str(self.scheduler_job._generate_zombie_message_details(ti))
             assert requests[0].is_failure_callback is True
             assert isinstance(requests[0].simple_task_instance, SimpleTaskInstance)
             assert ti.dag_id == requests[0].simple_task_instance.dag_id
@@ -4055,6 +4055,72 @@ def test_find_zombies(self):
             session.query(TaskInstance).delete()
             session.query(LocalTaskJob).delete()
 
+    def test_zombie_message(self):
+        """
+        Check that the zombie message comes out as expected
+        """
+
+        dagbag = DagBag(TEST_DAG_FOLDER, read_dags_from_db=False)
+        with create_session() as session:
+            session.query(LocalTaskJob).delete()
+            dag = dagbag.get_dag('example_branch_operator')
+            dag.sync_to_db()
+
+            dag_run = dag.create_dagrun(
+                state=DagRunState.RUNNING,
+                execution_date=DEFAULT_DATE,
+                run_type=DagRunType.SCHEDULED,
+                session=session,
+            )
+
+            self.scheduler_job = SchedulerJob(subdir=os.devnull)
+            self.scheduler_job.executor = MockExecutor()
+            self.scheduler_job.processor_agent = mock.MagicMock()
+
+            # We will provision 2 tasks so we can check we only find zombies from this scheduler
+            tasks_to_setup = ['branching', 'run_this_first']
+
+            for task_id in tasks_to_setup:
+                task = dag.get_task(task_id=task_id)
+                ti = TaskInstance(task, run_id=dag_run.run_id, state=State.RUNNING)
+                ti.queued_by_job_id = 999
+
+                local_job = LocalTaskJob(ti)
+                local_job.state = State.SHUTDOWN
+
+                session.add(local_job)
+                session.flush()
+
+                ti.job_id = local_job.id
+                session.add(ti)
+                session.flush()
+
+            assert task.task_id == 'run_this_first'  # Make sure we have the task/ti we expect
+
+            ti.queued_by_job_id = self.scheduler_job.id
+            session.flush()
+
+            zombie_message = self.scheduler_job._generate_zombie_message_details(ti)
+            assert zombie_message == {
+                'DAG Id': 'example_branch_operator',
+                'Task Id': 'run_this_first',
+                'Run Id': 'scheduled__2016-01-01T00:00:00+00:00',
+            }
+
+            ti.hostname = "10.10.10.10"
+            ti.map_index = 2
+            ti.external_executor_id = "abcdefg"
+
+            zombie_message = self.scheduler_job._generate_zombie_message_details(ti)
+            assert zombie_message == {
+                'DAG Id': 'example_branch_operator',
+                'Task Id': 'run_this_first',
+                'Run Id': 'scheduled__2016-01-01T00:00:00+00:00',
+                "Hostname": "10.10.10.10",
+                "Map Index": 2,
+                "External Executor Id": "abcdefg",
+            }
+
     def test_find_zombies_handle_failure_callbacks_are_correctly_passed_to_dag_processor(self):
         """
         Check that the same set of failure callback with zombies are passed to the dag