Copilot commented on code in PR #65002:
URL: https://github.com/apache/airflow/pull/65002#discussion_r3097470593
##########
task-sdk/tests/task_sdk/execution_time/test_supervisor.py:
##########
@@ -3147,6 +3147,22 @@ def
test_process_log_messages_from_subprocess(monkeypatch, caplog):
]
+def test_process_log_messages_tolerates_closed_log_handle():
+ """ValueError from a closed log handle must not propagate out of the
generator."""
+ broken_logger = MagicMock()
+ broken_logger.log.side_effect = ValueError("write to closed file")
+
+ with mock.patch(
+ "airflow.sdk.execution_time.supervisor.reconfigure_logger",
+ return_value=broken_logger,
+ ):
+ gen = process_log_messages_from_subprocess(loggers=(broken_logger,))
+ next(gen)
Review Comment:
New test creates a `MagicMock` logger without a `spec`/`autospec`, which can
let attribute/method typos slip through and reduces the value of the test.
Prefer `mocker.Mock(spec=FilteringBoundLogger)`/`mocker.create_autospec(...)`
(or a real `structlog.get_logger()` with a stubbed underlying logger) so the
mocked interface matches the real logger API.
##########
task-sdk/src/airflow/sdk/execution_time/supervisor.py:
##########
@@ -1923,7 +1923,11 @@ def process_log_messages_from_subprocess(
if level := NAME_TO_LEVEL.get(event.pop("level")):
msg = event.pop("event", None)
for target in loggers:
- target.log(level, msg, **event)
+ try:
+ target.log(level, msg, **event)
+ except ValueError:
+ # Log handle already closed; discard remaining output from
killed subprocess.
+ return
Review Comment:
Catching `ValueError` from `target.log()` and immediately `return` will
silently stop log draining for *all* targets and will also swallow unrelated
`ValueError`s (e.g. a formatting/processor bug), making those failures very
hard to detect. Consider narrowing the exception handling to the specific
"write to closed file" case (or otherwise re-raising unexpected `ValueError`s),
and optionally skipping only the failing logger instead of terminating the
generator entirely.
##########
airflow-core/tests/unit/dag_processing/test_manager.py:
##########
@@ -831,6 +831,79 @@ def test_kill_timed_out_processors_no_kill(self):
manager._kill_timed_out_processors()
mock_kill.assert_not_called()
+ def test_terminate_orphan_processes_deregisters_sockets(self):
+ manager = DagFileProcessorManager(max_runs=1)
+ processor, _ = self.mock_processor()
+
+ sock1, sock2 = MagicMock(), MagicMock()
+ processor._open_sockets[sock1] = "log"
+ processor._open_sockets[sock2] = "log"
+
+ file_info = DagFileInfo(
+ bundle_name="testing", rel_path=Path("removed.py"),
bundle_path=TEST_DAGS_FOLDER
+ )
+ manager._processors = {file_info: processor}
+ manager.selector = MagicMock()
+
+ with mock.patch.object(type(processor), "kill"):
+ manager.terminate_orphan_processes(present=set())
+
+ manager.selector.unregister.assert_any_call(sock1)
+ manager.selector.unregister.assert_any_call(sock2)
+ sock1.close.assert_called_once()
+ sock2.close.assert_called_once()
+ assert not processor._open_sockets
+ processor.logger_filehandle.close.assert_called_once()
+
+ def test_kill_timed_out_processors_deregisters_sockets(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)
+
+ sock = MagicMock()
+ processor._open_sockets[sock] = "log"
+
Review Comment:
These tests add `MagicMock()` sockets used as keys in `_open_sockets`
without any `spec`/`autospec`. Using a spec (e.g. `socket.socket`) would better
reflect real behavior (notably `fileno()`/`close()` semantics) and make the
test more robust against API drift.
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -968,6 +968,15 @@ def remove_orphaned_file_stats(self, present:
set[DagFileInfo]):
for file in stats_to_remove:
del self._file_stats[file]
+ def _deregister_processor_sockets(self, processor:
DagFileProcessorProcess) -> None:
+ """Unregister and close all sockets for a killed processor before
closing its log handle."""
+ for sock in list(processor._open_sockets.keys()):
+ with contextlib.suppress(KeyError):
+ self.selector.unregister(sock)
+ with contextlib.suppress(Exception):
Review Comment:
`_deregister_processor_sockets()` uses `self.selector.unregister(sock)` and
then suppresses all exceptions from `sock.close()`. For correctness and
future-proofing, it should operate on `processor.selector` (the selector the
sockets were registered with) and avoid suppressing unexpected exceptions—limit
suppression to expected cases (e.g. `KeyError` for unregister, and
`OSError`/`ValueError` for close). Swallowing arbitrary exceptions here can
hide real bugs and potentially leave sockets registered/open in production.
```suggestion
processor.selector.unregister(sock)
with contextlib.suppress(OSError, ValueError):
```
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
