This is an automated email from the ASF dual-hosted git repository.
vatsrahul1001 pushed a commit to branch v3-3-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-3-test by this push:
new 3b6f63eef97 Fix DagProcessorJob crash on orphan processor kill
(#69523) (#72888)
3b6f63eef97 is described below
commit 3b6f63eef972193942295619115a6fc0c6ba35fc
Author: Rahul Vats <[email protected]>
AuthorDate: Thu Sep 10 22:18:16 2026 +0530
Fix DagProcessorJob crash on orphan processor kill (#69523) (#72888)
After a processor is SIGKILLed its sockets stay registered on the shared
selector, so the next select() callback writes to an already-closed log
handle and the job dies with "ValueError: write to closed file".
Add WatchedSubprocess.cleanup_sockets_after_kill() to drain the log-bearing
sockets, run their close callbacks and unregister them, and call it from
DagFileProcessorProcess.close(). Request sockets are closed without being
drained so no request frames get dispatched after the kill.
Also route supervisor log writes through _log_to_target() so a logger whose
handle is already closed is skipped instead of taking the process down.
closes: #64959
(cherry picked from commit 6d9cc2ed37b0926e6e7ce85677152ff90f4e02da)
Co-authored-by: Hemkumar Chheda <[email protected]>
---
airflow-core/src/airflow/dag_processing/manager.py | 6 +-
.../src/airflow/dag_processing/processor.py | 14 ++-
.../tests/unit/dag_processing/test_manager.py | 67 ++++++++++++
.../src/airflow/sdk/execution_time/supervisor.py | 55 +++++++++-
.../task_sdk/execution_time/test_supervisor.py | 116 +++++++++++++++++++++
5 files changed, 251 insertions(+), 7 deletions(-)
diff --git a/airflow-core/src/airflow/dag_processing/manager.py
b/airflow-core/src/airflow/dag_processing/manager.py
index 6d05c8d0a56..b1a18c14169 100644
--- a/airflow-core/src/airflow/dag_processing/manager.py
+++ b/airflow-core/src/airflow/dag_processing/manager.py
@@ -1234,7 +1234,7 @@ class DagFileProcessorManager(LoggingMixin):
),
)
processor.kill(signal.SIGKILL)
- processor.logger_filehandle.close()
+ processor.close()
self._file_stats.pop(file, None)
@provide_session
@@ -1365,7 +1365,7 @@ class DagFileProcessorManager(LoggingMixin):
for file in finished:
processor = self._processors.pop(file)
- processor.logger_filehandle.close()
+ processor.close()
def _get_log_dir(self) -> str:
return os.path.join(self.base_log_dir,
timezone.utcnow().strftime("%Y-%m-%d"))
@@ -1671,7 +1671,7 @@ class DagFileProcessorManager(LoggingMixin):
# Clean up `self._processors` after iterating over it
for proc in processors_to_remove:
processor = self._processors.pop(proc)
- processor.logger_filehandle.close()
+ processor.close()
def _add_files_to_queue(
self,
diff --git a/airflow-core/src/airflow/dag_processing/processor.py
b/airflow-core/src/airflow/dag_processing/processor.py
index 89b5243d762..9f8f18716bc 100644
--- a/airflow-core/src/airflow/dag_processing/processor.py
+++ b/airflow-core/src/airflow/dag_processing/processor.py
@@ -92,6 +92,7 @@ from airflow.serialization.serialized_objects import
DagSerialization, LazyDeser
from airflow.utils.dag_version_inflation_checker import
check_dag_file_stability
from airflow.utils.file import iter_airflow_imports
from airflow.utils.helpers import prune_dict
+from airflow.utils.log.logging_mixin import LoggingMixin
from airflow.utils.state import TaskInstanceState
if TYPE_CHECKING:
@@ -573,7 +574,7 @@ def in_process_api_server() -> InProcessExecutionAPI:
@attrs.define(kw_only=True)
-class DagFileProcessorProcess(WatchedSubprocess):
+class DagFileProcessorProcess(WatchedSubprocess, LoggingMixin):
"""
Parses dags with Task SDK API.
@@ -742,3 +743,14 @@ class DagFileProcessorProcess(WatchedSubprocess):
def wait(self) -> int:
raise NotImplementedError(f"Don't call wait on {type(self).__name__}
objects")
+
+ def close(self):
+ self.cleanup_sockets_after_kill()
+ try:
+ self.logger_filehandle.close()
+ except OSError:
+ self.log.warning(
+ "Failed to close log file handle for %s",
+ self.dag_file_rel_path,
+ exc_info=True,
+ )
diff --git a/airflow-core/tests/unit/dag_processing/test_manager.py
b/airflow-core/tests/unit/dag_processing/test_manager.py
index c5e09de0563..64cb734e967 100644
--- a/airflow-core/tests/unit/dag_processing/test_manager.py
+++ b/airflow-core/tests/unit/dag_processing/test_manager.py
@@ -23,6 +23,7 @@ import logging
import os
import random
import re
+import selectors
import shutil
import signal
import textwrap
@@ -1421,6 +1422,72 @@ class TestDagFileProcessorManager:
_, kwargs = mock_start.call_args
assert kwargs["subprocess_logs_to_stdout"] is
expected_subprocess_logs_to_stdout
+ def test_terminate_orphan_processes_kills_then_closes_processor(self):
+ manager = DagFileProcessorManager(max_runs=1)
+ processor, _ = self.mock_processor()
+ file_info = DagFileInfo(
+ bundle_name="testing", rel_path=Path("removed.py"),
bundle_path=TEST_DAGS_FOLDER
+ )
+ manager._processors = {file_info: processor}
+
+ call_order: list[str] = []
+ processor.close = mock.Mock(side_effect=lambda:
call_order.append("close"))
+
+ with mock.patch.object(
+ type(processor), "kill", side_effect=lambda *_args, **_kwargs:
call_order.append("kill")
+ ):
+ manager.terminate_orphan_processes(present=set())
+
+ assert call_order == ["kill", "close"]
+
+ def
test_terminate_orphan_processes_does_not_dispatch_request_frames_after_kill(self):
+ manager = DagFileProcessorManager(max_runs=1)
+ processor, _ = self.mock_processor()
+ request_sock, request_peer = socketpair()
+ real_selector = selectors.DefaultSelector()
+ try:
+ processor.selector = real_selector
+ processor._open_sockets[request_sock] = "requests"
+
+ file_info = DagFileInfo(
+ bundle_name="testing", rel_path=Path("removed.py"),
bundle_path=TEST_DAGS_FOLDER
+ )
+ manager._processors = {file_info: processor}
+
+ request_handler = mock.Mock(return_value=False)
+
+ def on_close(sock):
+ real_selector.unregister(sock)
+
+ real_selector.register(request_sock, selectors.EVENT_READ,
(request_handler, on_close))
+
+ with mock.patch.object(type(processor), "kill"):
+ manager.terminate_orphan_processes(present=set())
+
+ request_handler.assert_not_called()
+ with pytest.raises((KeyError, ValueError)):
+ real_selector.get_key(request_sock)
+ finally:
+ real_selector.close()
+ request_peer.close()
+
+ def test_kill_timed_out_processors_kills_then_closes_processor(self):
+ manager = DagFileProcessorManager(max_runs=1, processor_timeout=5)
+ start_time = time.monotonic() - manager.processor_timeout - 1
+ processor, _ = self.mock_processor(start_time=start_time)
+ file_info = DagFileInfo(bundle_name="testing",
rel_path=Path("abc.txt"), bundle_path=TEST_DAGS_FOLDER)
+ manager._processors = {file_info: processor}
+
+ call_order: list[str] = []
+ processor.close = mock.Mock(side_effect=lambda:
call_order.append("close"))
+
+ with mock.patch.object(
+ type(processor), "kill", side_effect=lambda *_args, **_kwargs:
call_order.append("kill")
+ ):
+ manager._kill_timed_out_processors()
+
+ assert call_order == ["kill", "close"]
+
def test_kill_timed_out_processors_kill(self):
manager = DagFileProcessorManager(max_runs=1, processor_timeout=5)
# Set start_time to ensure timeout occurs: start_time = current_time -
(timeout + 1) = always (timeout + 1) seconds
diff --git a/task-sdk/src/airflow/sdk/execution_time/supervisor.py
b/task-sdk/src/airflow/sdk/execution_time/supervisor.py
index db1e3488d36..abec13814ae 100644
--- a/task-sdk/src/airflow/sdk/execution_time/supervisor.py
+++ b/task-sdk/src/airflow/sdk/execution_time/supervisor.py
@@ -1033,6 +1033,46 @@ class WatchedSubprocess:
pass
self._process.send_signal(sig)
+ def cleanup_sockets_after_kill(self) -> None:
+ """Drain log-bearing sockets, then close every remaining socket after
a forced kill."""
+ for sock, socket_type in list(self._open_sockets.items()):
+ try:
+ key = self.selector.get_key(sock)
+ except KeyError:
+ key = None
+
+ if key is not None:
+ socket_handler, on_close = key.data
+ try:
+ if socket_type != "requests":
+ sock.setblocking(False)
+ while True:
+ try:
+ if not socket_handler(sock):
+ break
+ except (BlockingIOError, InterruptedError,
OSError):
+ break
+
+ if on_close is not None:
+ on_close(sock)
+ else:
+ with suppress(KeyError):
+ self.selector.unregister(sock)
+ self._open_sockets.pop(sock, None)
+ except Exception:
+ log.exception(
+ "Failed to clean up killed subprocess socket",
+ pid=self.pid,
+ socket_type=socket_type,
+ )
+ with suppress(KeyError):
+ self.selector.unregister(sock)
+ self._open_sockets.pop(sock, None)
+ with suppress(OSError, ValueError):
+ sock.close()
+
+ self._open_sockets.clear()
+
def kill(
self,
signal_to_send: signal.Signals = signal.SIGINT,
@@ -2405,9 +2445,18 @@ def process_log_messages_from_subprocess(
event["error_detail"] = exc
if level := NAME_TO_LEVEL.get(event.pop("level")):
- msg = event.pop("event", None)
+ msg = event.pop("event", None) or ""
for target in loggers:
- target.log(level, msg, **event)
+ _log_to_target(target, level, msg, **event)
+
+
+def _log_to_target(target: FilteringBoundLogger, level: int, msg: str,
**event) -> None:
+ try:
+ target.log(level, msg, **event)
+ except ValueError as e:
+ if "closed file" not in str(e):
+ raise
+ log.debug("Dropped log line for closed logger handle", level=level,
logger=event.get("logger"))
def forward_to_log(
@@ -2422,7 +2471,7 @@ def forward_to_log(
except UnicodeDecodeError:
msg = line.decode("ascii", errors="replace")
for log in target_loggers:
- log.log(level, msg, logger=logger)
+ _log_to_target(log, level, msg, logger=logger)
def ensure_secrets_backend_loaded() -> list[BaseSecretsBackend]:
diff --git a/task-sdk/tests/task_sdk/execution_time/test_supervisor.py
b/task-sdk/tests/task_sdk/execution_time/test_supervisor.py
index f79e10e364a..f12d2faa801 100644
--- a/task-sdk/tests/task_sdk/execution_time/test_supervisor.py
+++ b/task-sdk/tests/task_sdk/execution_time/test_supervisor.py
@@ -50,6 +50,7 @@ from opentelemetry.sdk.trace.export import SimpleSpanProcessor
from opentelemetry.sdk.trace.export.in_memory_span_exporter import
InMemorySpanExporter
from opentelemetry.trace import get_current_span
from pytest_unordered import unordered
+from structlog.typing import FilteringBoundLogger
from task_sdk import FAKE_BUNDLE, make_client
from uuid6 import uuid7
@@ -169,6 +170,7 @@ from airflow.sdk.execution_time.supervisor import (
WatchedSubprocess,
_make_process_nondumpable,
_remote_logging_conn,
+ forward_to_log,
in_process_api_server,
make_buffered_socket_reader,
process_log_messages_from_subprocess,
@@ -4136,6 +4138,120 @@ def
test_process_log_messages_from_subprocess(monkeypatch, caplog):
]
[email protected](
+ "error_message",
+ ["write to closed file", "I/O operation on closed file"],
+)
+def test_process_log_messages_closed_logger_is_skipped(error_message):
+ closed_logger = mock.Mock(spec=FilteringBoundLogger)
+ closed_logger.log.side_effect = ValueError(error_message)
+
+ good_logger = mock.Mock(spec=FilteringBoundLogger)
+
+ def fake_reconfigure(logger, *args, **kwargs):
+ return logger
+
+ with (
+ mock.patch(
+ "airflow.sdk.execution_time.supervisor.reconfigure_logger",
+ side_effect=fake_reconfigure,
+ ),
+ mock.patch.object(supervisor.log, "debug") as mock_debug,
+ ):
+ gen = process_log_messages_from_subprocess(loggers=(closed_logger,
good_logger))
+ next(gen)
+
+ gen.send(b'{"level": "info", "event": "hello"}\n')
+ gen.send(b'{"level": "info", "event": "world"}\n')
+
+ assert good_logger.log.call_count == 2
+ assert mock_debug.call_count == 2
+
+
+def test_forward_to_log_closed_logger_is_skipped():
+ closed_logger = mock.Mock(spec=FilteringBoundLogger)
+ closed_logger.log.side_effect = ValueError("I/O operation on closed file")
+ good_logger = mock.Mock(spec=FilteringBoundLogger)
+
+ with mock.patch.object(supervisor.log, "debug") as mock_debug:
+ gen = forward_to_log((closed_logger, good_logger),
logger="task.stdout", level=logging.INFO)
+ next(gen)
+ gen.send(b"hello\n")
+ gen.send(b"world\n")
+
+ assert good_logger.log.call_count == 2
+ good_logger.log.assert_any_call(logging.INFO, "hello",
logger="task.stdout")
+ good_logger.log.assert_any_call(logging.INFO, "world",
logger="task.stdout")
+ assert mock_debug.call_count == 2
+
+
+def test_process_log_messages_unexpected_value_error_is_reraised():
+ """A ValueError unrelated to a closed file handle must propagate, not be
silently swallowed."""
+ buggy_logger = mock.Mock(spec=FilteringBoundLogger)
+ buggy_logger.log.side_effect = ValueError("unexpected formatting bug")
+
+ def fake_reconfigure(log, *args, **kwargs):
+ return log
+
+ with mock.patch(
+ "airflow.sdk.execution_time.supervisor.reconfigure_logger",
+ side_effect=fake_reconfigure,
+ ):
+ gen = process_log_messages_from_subprocess(loggers=(buggy_logger,))
+ next(gen)
+
+ with pytest.raises(ValueError, match="unexpected formatting bug"):
+ gen.send(b'{"level": "info", "event": "test"}\n')
+
+
+def test_cleanup_sockets_after_kill_drains_logs_but_not_requests(mocker):
+ request_read, request_write = socket.socketpair()
+ stdout_read, stdout_write = socket.socketpair()
+ log_read, log_write = socket.socketpair()
+
+ subprocess = ActivitySubprocess(
+ process_log=mocker.MagicMock(),
+ id=TI_ID,
+ pid=12345,
+ stdin=stdout_write,
+ client=mocker.Mock(),
+ process=mocker.Mock(),
+ )
+ selector = selectors.DefaultSelector()
+ subprocess.selector = selector
+
+ request_handler = mock.Mock(return_value=False)
+ stdout_handler = mock.Mock(return_value=False)
+ log_handler = mock.Mock(return_value=False)
+
+ def on_close(sock):
+ selector.unregister(sock)
+ subprocess._open_sockets.pop(sock, None)
+
+ try:
+ subprocess._open_sockets[request_read] = "requests"
+ subprocess._open_sockets[stdout_read] = "stdout"
+ subprocess._open_sockets[log_read] = "logs"
+
+ selector.register(request_read, selectors.EVENT_READ,
(request_handler, on_close))
+ selector.register(stdout_read, selectors.EVENT_READ, (stdout_handler,
on_close))
+ selector.register(log_read, selectors.EVENT_READ, (log_handler,
on_close))
+
+ subprocess.cleanup_sockets_after_kill()
+
+ request_handler.assert_not_called()
+ stdout_handler.assert_called_once_with(stdout_read)
+ log_handler.assert_called_once_with(log_read)
+ assert not subprocess._open_sockets
+ with pytest.raises((KeyError, ValueError)):
+ selector.get_key(request_read)
+ finally:
+ selector.close()
+ request_write.close()
+ stdout_write.close()
+ log_write.close()
+
+
def test_reinit_supervisor_comms(monkeypatch, client_with_ti_start, caplog):
def subprocess_main():
# This is run in the subprocess!
