Fix processor cleanup

References to processors weren't being cleaned up after killing them in the event of a timeout. This lead to a crash caused by an unhandled exception when trying to read from a closed end of a pipe.
apache · potiuk · Apr 6, 2022 · Apr 1, 2022 · Apr 1, 2022 · Apr 2, 2022
commit 85f5440441dc3ee06f8748b5c131616c65457fd2
diff --git a/airflow/dag_processing/manager.py b/airflow/dag_processing/manager.py
@@ -1065,6 +1065,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:
@@ -1080,6 +1081,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.