This is an automated email from the ASF dual-hosted git repository.
jason810496 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 87089a53a2e Fix crash when tailing logs of a running task instance
(#69503)
87089a53a2e is described below
commit 87089a53a2e2358e96defae35979b8d271375914
Author: Jason(Zhe-You) Liu <[email protected]>
AuthorDate: Tue Jul 7 14:44:30 2026 +0900
Fix crash when tailing logs of a running task instance (#69503)
FileTaskHandler._read() returns an itertools.islice stream when
resuming a log read via metadata["log_pos"] (introduced in #63531),
but _is_logs_stream_like() only recognized chain and generator
objects, so the public read() wrapper raised
TypeError: Invalid log stream type... Got islice instead
whenever a client polled or tailed a running task's logs.
---
.../src/airflow/utils/log/file_task_handler.py | 2 +-
airflow-core/tests/unit/utils/test_log_handlers.py | 39 ++++++++++++++++++++++
2 files changed, 40 insertions(+), 1 deletion(-)
diff --git a/airflow-core/src/airflow/utils/log/file_task_handler.py
b/airflow-core/src/airflow/utils/log/file_task_handler.py
index 9eec24b1813..81f9a89c458 100644
--- a/airflow-core/src/airflow/utils/log/file_task_handler.py
+++ b/airflow-core/src/airflow/utils/log/file_task_handler.py
@@ -398,7 +398,7 @@ def _interleave_logs(*log_streams: RawLogStream) ->
StructuredLogStream:
def _is_logs_stream_like(log) -> bool:
"""Check if the logs are stream-like."""
- return isinstance(log, (chain, GeneratorType))
+ return isinstance(log, (chain, islice, GeneratorType))
def _get_compatible_log_stream(
diff --git a/airflow-core/tests/unit/utils/test_log_handlers.py
b/airflow-core/tests/unit/utils/test_log_handlers.py
index 24a2e367a38..8248e57e4d5 100644
--- a/airflow-core/tests/unit/utils/test_log_handlers.py
+++ b/airflow-core/tests/unit/utils/test_log_handlers.py
@@ -504,6 +504,31 @@ class TestFileTaskLogHandler:
assert extract_events(log_handler_output_stream) == ["line 3"]
assert metadata == {"end_of_log": True, "log_pos": 3}
+
@patch("airflow.utils.log.file_task_handler.FileTaskHandler._read_from_local")
+ def test_read_respects_log_pos_metadata(self, mock_read_local,
create_task_instance):
+ """The public `read()` wrapper must accept the `islice` stream
`_read()` returns for log_pos reads."""
+ mock_read_local.return_value = (
+ ["the messages"],
+ [convert_list_to_stream(["line 1", "line 2", "line 3"])],
+ )
+ local_log_file_read = create_task_instance(
+ dag_id="dag_for_testing_local_log_read",
+ task_id="task_for_testing_local_log_read",
+ run_type=DagRunType.SCHEDULED,
+ logical_date=DEFAULT_DATE,
+ )
+ fth = FileTaskHandler("")
+
+ log_handler_output_stream, metadata = fth.read(
+ local_log_file_read,
+ try_number=1,
+ metadata={"log_pos": 2},
+ )
+
+ # Should resume from the third line only.
+ assert extract_events(log_handler_output_stream) == ["line 3"]
+ assert metadata == {"end_of_log": True, "log_pos": 3}
+
def test__read_from_local(self, tmp_path):
"""Tests the behavior of method _read_from_local"""
path1 = tmp_path / "hello1.log"
@@ -1179,6 +1204,20 @@ def test__is_sort_key_with_default_timestamp(timestamp,
line_num, expected):
True,
id="chain_log_stream",
),
+ pytest.param(
+ itertools.islice(
+ convert_list_to_stream(
+ [
+ "2022-11-16T00:05:54.278000-08:00",
+ "2022-11-16T00:05:54.457000-08:00",
+ ]
+ ),
+ 1,
+ None,
+ ),
+ True,
+ id="islice_log_stream",
+ ),
pytest.param(
[
"2022-11-16T00:05:54.278000-08:00",
