Fix processor cleanup on DagFileProcessorManager

airflow/dag_processing/manager.py

@@ -1058,6 +1058,7 @@ def prepare_file_path_queue(self):
     def _kill_timed_out_processors(self):
         """Kill any file processors that timeout to defend against process hangs."""
         now = timezone.utcnow()
+        processors_to_remove = []
         for file_path, processor in self._processors.items():
             duration = now - processor.start_time
             if duration > self._processor_timeout:
@@ -1073,6 +1074,14 @@ def _kill_timed_out_processors(self):
                 Stats.incr('dag_file_processor_timeouts')
                 processor.kill()
 
+                # Clean up processor references
+                self.waitables.pop(processor.waitable_handle)
+                processors_to_remove.append(file_path)
+
+        # Clean up `self._processors` after iterating over it
+        for proc in processors_to_remove:
+            self._processors.pop(proc)
+
     def max_runs_reached(self):
         """:return: whether all file paths have been processed max_runs times"""
         if self._max_runs == -1:  # Unlimited runs.

airflow/dag_processing/processor.py

@@ -21,6 +21,7 @@
 import os
 import signal
 import threading
+import time
 from contextlib import redirect_stderr, redirect_stdout, suppress
 from datetime import timedelta
 from multiprocessing.connection import Connection as MultiprocessingConnection
@@ -231,6 +232,12 @@ def _kill_process(self) -> None:
         if self._process.is_alive() and self._process.pid:
             self.log.warning("Killing DAGFileProcessorProcess (PID=%d)", self._process.pid)
             os.kill(self._process.pid, signal.SIGKILL)
+
+            # Reap the spawned zombie. We active wait, because in Python 3.9 `waitpid` might lead to an
+            # exception, due to change in Python standard library and possibility of race condition
+            # see https://bugs.python.org/issue42558
+            while self._process.poll() is None:
+                time.sleep(0.001)
         if self._parent_channel:
             self._parent_channel.close()