(airflow) branch main updated: Resolve handler close race condition at triggerer shutdown (#37206)

[je...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

jedcunningham pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git


The following commit(s) were added to refs/heads/main by this push:
     new e24f0a975d Resolve handler close race condition at triggerer shutdown (#37206)
e24f0a975d is described below

commit e24f0a975d070cee4b376da7d86fe22d6b02502c
Author: Daniel Standish <15...@users.noreply.github.com>
AuthorDate: Tue Feb 6 15:11:01 2024 -0800

    Resolve handler close race condition at triggerer shutdown (#37206)
    
    At shutdown, a trigger may emit a "close" signal around the same time as logging.shutDown is called, so the handler may have been removed from the handlers by one thread when the other tries to delete it.
---
 airflow/utils/log/trigger_handler.py | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

diff --git a/airflow/utils/log/trigger_handler.py b/airflow/utils/log/trigger_handler.py
index aa9a43ec87..02af5f306f 100644
--- a/airflow/utils/log/trigger_handler.py
+++ b/airflow/utils/log/trigger_handler.py
@@ -19,6 +19,7 @@ from __future__ import annotations
 
 import asyncio
 import logging
+from contextlib import suppress
 from contextvars import ContextVar
 from copy import copy
 from logging.handlers import QueueHandler
@@ -110,7 +111,8 @@ class TriggererHandlerWrapper(logging.Handler):
         h = self.handlers.get(trigger_id)
         if h:
             h.close()
-            del self.handlers[trigger_id]
+            with suppress(KeyError):  # race condition between `handle` and `close`
+                del self.handlers[trigger_id]
 
     def flush(self):
         for h in self.handlers.values():
@@ -118,9 +120,7 @@ class TriggererHandlerWrapper(logging.Handler):
 
     def close(self):
         for trigger_id in list(self.handlers.keys()):
-            h = self.handlers[trigger_id]
-            h.close()
-            del self.handlers[trigger_id]
+            self.close_one(trigger_id)
 
 
 class LocalQueueHandler(QueueHandler):