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 9eec24b18134e..81f9a89c45865 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 24a2e367a3854..8248e57e4d538 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 @@ def test__read_when_local_respects_log_pos_metadata(self, mock_read_local, creat 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",