fix logic to use close method instead python __del__

Author: wjddn279

airflow-core/src/airflow/jobs/triggerer_job_runner.py

@@ -322,9 +322,9 @@ def __call__(self, processors: Iterable[structlog.typing.Processor]) -> WrappedL
         self.bound_logger = logger
         return logger
 
-    def __del__(self):
-        # Explicitly close the file descriptor when the logger is garbage collected.
-        if hasattr(self, "_filehandle") and self._filehandle:
+    def close(self):
+        # Explicitly close the file descriptor.
+        if hasattr(self, "_filehandle") and self._filehandle and not self._filehandle.closed:
             self._filehandle.close()
 
     def upload_to_remote(self):
@@ -433,6 +433,7 @@ def _handle_request(self, msg: ToTriggerSupervisor, log: FilteringBoundLogger, r
                 # only need to remove the last reference to it to close the open FH
                 if factory := self.logger_cache.pop(id, None):
                     factory.upload_to_remote()
+                    factory.close()
 
             response = messages.TriggerStateSync(
                 to_create=[],

airflow-core/tests/unit/jobs/test_triggerer_job.py

@@ -42,6 +42,7 @@
     ToTriggerSupervisor,
     TriggerCommsDecoder,
     TriggererJobRunner,
+    TriggerLoggingFactory,
     TriggerRunner,
     TriggerRunnerSupervisor,
     messages,
@@ -303,6 +304,19 @@ def test_trigger_log(mock_monotonic, trigger, watcher_count, trigger_count, sess
     trigger_runner_supervisor.kill(force=False)
 
 
+def test_trigger_logger_close():
+    logger = TriggerLoggingFactory(log_path="/tmp/test.log", ti=MagicMock())
+
+    mock_fh = MagicMock()
+    mock_fh.closed = False
+
+    logger._filehandle = mock_fh
+
+    logger.close()
+
+    mock_fh.close.assert_called_once()
+
+
 def test_trigger_logger_fd_closed_when_removed(session):
 
     trigger = TimeDeltaTrigger(datetime.timedelta(seconds=0.5))