[GitHub] [airflow] yuqian90 commented on a change in pull request #5605: [AIRFLOW-4883] Kill hung file process managers

[GitBox <gi...@apache.org>]

yuqian90 commented on a change in pull request #5605: [AIRFLOW-4883] Kill hung file process managers
URL: https://github.com/apache/airflow/pull/5605#discussion_r305567080
 
 

 ##########
 File path: airflow/utils/dag_processing.py
 ##########
 @@ -1229,6 +1243,21 @@ def heartbeat(self):
 
         return simple_dags
 
+    def _kill_timed_out_processors(self):
+        """
+        Kill any file processors that timeout to defend against process hangs.
+        """
+        now = timezone.utcnow()
+        for file_path, processor in self._processors.items():
+            duration = now - processor.start_time
+            if duration > self._processor_timeout:
+                self.log.info(
+                    "Processor for %s with PID %s started at %s has timed out, "
+                    "killing it.",
+                    processor.file_path, processor.pid, processor.start_time.isoformat())
+                Stats.incr('dag_file_processor_timeouts', 1, 1)
+                processor.kill()
 
 Review comment:
   This line seems to cause scheduler to crash. There is no `.kill()`. Do you mean `.terminate()` ?
   
   ```
     File "/usr/local/lib/python3.6/multiprocessing/process.py", line 258, in _bootstrap
       self.run()
     File "/usr/local/lib/python3.6/multiprocessing/process.py", line 93, in run
       self._target(*self._args, **self._kwargs)
     File "/opt/airflow/airflow/utils/dag_processing.py", line 581, in helper
       processor_manager.start()
     File "/opt/airflow/airflow/utils/dag_processing.py", line 825, in start
       self.start_in_sync()
     File "/opt/airflow/airflow/utils/dag_processing.py", line 891, in start_in_sync
       simple_dags = self.heartbeat()
     File "/opt/airflow/airflow/utils/dag_processing.py", line 1153, in heartbeat
       self._kill_timed_out_processors()
     File "/opt/airflow/airflow/utils/dag_processing.py", line 1259, in _kill_timed_out_processors
       processor.kill()
   AttributeError: 'DagFileProcessor' object has no attribute 'kill'
   ```

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services