(airflow) branch v3-0-test updated: [v3-0-test] Explicitly close log file descriptor in the supervise function (#51627) (#51654)

[kaxilnaik]

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

kaxilnaik pushed a commit to branch v3-0-test
in repository https://gitbox.apache.org/repos/asf/airflow.git


The following commit(s) were added to refs/heads/v3-0-test by this push:
     new 86397cfdf0d [v3-0-test] Explicitly close log file descriptor in the 
supervise function (#51627) (#51654)
86397cfdf0d is described below

commit 86397cfdf0d756f339f08e1d889a3374c8c793c5
Author: github-actions[bot] 
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Thu Jun 12 20:39:03 2025 +0530

    [v3-0-test] Explicitly close log file descriptor in the supervise function 
(#51627) (#51654)
    
    We didn't close log file descriptor properly hence leading to too many open 
files error from the operating system.
    (cherry picked from commit 4a0a89b6733002b8adccaf09d9f76d26c0d9d706)
    
    Co-authored-by: Adylzhan Khashtamov <adil.khashta...@gmail.com>
---
 task-sdk/src/airflow/sdk/execution_time/supervisor.py | 10 ++++++++--
 1 file changed, 8 insertions(+), 2 deletions(-)

diff --git a/task-sdk/src/airflow/sdk/execution_time/supervisor.py 
b/task-sdk/src/airflow/sdk/execution_time/supervisor.py
index b2dd76d85b1..f3cc633fb63 100644
--- a/task-sdk/src/airflow/sdk/execution_time/supervisor.py
+++ b/task-sdk/src/airflow/sdk/execution_time/supervisor.py
@@ -35,6 +35,7 @@ from http import HTTPStatus
 from socket import SO_SNDBUF, SOL_SOCKET, SocketIO, socket, socketpair
 from typing import (
     TYPE_CHECKING,
+    BinaryIO,
     Callable,
     ClassVar,
     NoReturn,
@@ -1512,6 +1513,7 @@ def supervise(
 
     # TODO: Use logging providers to handle the chunked upload for us etc.
     logger: FilteringBoundLogger | None = None
+    log_file_descriptor: BinaryIO | TextIO | None = None
     if log_path:
         # If we are told to write logs to a file, redirect the task logger to 
it. Make sure we append to the
         # file though, otherwise when we resume we would lose the logs from 
the start->deferral segment if it
@@ -1522,9 +1524,11 @@ def supervise(
 
         pretty_logs = False
         if pretty_logs:
-            underlying_logger: WrappedLogger = 
structlog.WriteLogger(log_file.open("a", buffering=1))
+            log_file_descriptor = log_file.open("a", buffering=1)
+            underlying_logger: WrappedLogger = 
structlog.WriteLogger(cast("TextIO", log_file_descriptor))
         else:
-            underlying_logger = structlog.BytesLogger(log_file.open("ab"))
+            log_file_descriptor = log_file.open("ab")
+            underlying_logger = structlog.BytesLogger(cast("BinaryIO", 
log_file_descriptor))
         processors = logging_processors(enable_pretty_log=pretty_logs)[0]
         logger = structlog.wrap_logger(underlying_logger, 
processors=processors, logger_name="task").bind()
 
@@ -1549,4 +1553,6 @@ def supervise(
     exit_code = process.wait()
     end = time.monotonic()
     log.info("Task finished", exit_code=exit_code, duration=end - start, 
final_state=process.final_state)
+    if log_path and log_file_descriptor:
+        log_file_descriptor.close()
     return exit_code