fixup! Swap Dag Parsing to use the TaskSDK machinery.

apache · Dec 18, 2024 · e61c595 · e61c595
1 parent d4935ea
commit e61c595
Showing 1 changed file with 5 additions and 2 deletions.
diff --git a/airflow/dag_processing/manager.py b/airflow/dag_processing/manager.py
@@ -258,7 +258,10 @@ def terminate(self):
         """Send termination signal to DAG parsing processor manager to terminate all DAG file processors."""
         if self._process and self._process.is_alive():
             self.log.info("Sending termination message to manager.")
-            self._parent_signal_conn.send(None)
+            try:
+                self._parent_signal_conn.send(None)
+            except ConnectionError:
+                pass
             self._parent_signal_conn.close()
 
     def end(self):
@@ -321,7 +324,7 @@ class DagFileProcessorManager:
     stale_dag_threshold: float = attrs.field(factory=_config_int_factory("scheduler", "stale_dag_threshold"))
     last_dag_dir_refresh_time: float = attrs.field(default=0, init=False)
 
-    log: logging.Logger = log
+    log: logging.Logger = attrs.field(default=log, init=False)
 
     _last_deactivate_stale_dags_time: float = attrs.field(default=0, init=False)
     print_stats_interval: float = attrs.field(