jason810496 commented on code in PR #70671:
URL: https://github.com/apache/airflow/pull/70671#discussion_r4324800173
##########
airflow-core/tests/unit/utils/test_log_handlers.py:
##########
@@ -502,7 +502,34 @@ def test__read_when_local_respects_log_pos_metadata(self,
mock_read_local, creat
# 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}
+ assert metadata == {"end_of_log": True}
+
+ @pytest.mark.parametrize("ti_state", [TaskInstanceState.SUCCESS,
TaskInstanceState.FAILED])
Review Comment:
Good catch. Both new tests now set `ti.try_number = 1`, and the matrix
includes a running TI read at an older try, so each half of the `end_of_log`
check is covered. Updated in e4d36e30b1d.
##########
airflow-core/tests/unit/utils/test_log_handlers.py:
##########
@@ -502,7 +502,34 @@ def test__read_when_local_respects_log_pos_metadata(self,
mock_read_local, creat
# 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}
+ assert metadata == {"end_of_log": True}
Review Comment:
Parametrized `test_read_respects_log_pos_metadata` with a running TI on the
current try, which returns `["line 3"]` and `{"end_of_log": False, "log_pos":
3}`. Updated in e4d36e30b1d.
##########
airflow-core/src/airflow/utils/log/file_task_handler.py:
##########
@@ -647,21 +647,22 @@ def _read(
TaskInstanceState.DEFERRED,
)
- with LogStreamAccumulator(out_stream, HEAP_DUMP_SIZE) as
stream_accumulator:
- log_pos = stream_accumulator.total_lines
- out_stream = stream_accumulator.stream
-
- # skip log stream until the last position
- if metadata and "log_pos" in metadata:
- out_stream = islice(out_stream, metadata["log_pos"], None)
- else:
- # first time reading log, add messages before interleaved log
stream
- out_stream = chain(header, out_stream)
+ out_metadata: LogMetadata = {"end_of_log": end_of_log}
+ if not end_of_log:
Review Comment:
Agreed, the NDJSON path pays for a `log_pos` it never uses. I will handle it
in a follow-up where the caller says whether it needs `log_pos`, so only the
JSON branch of `get_log` asks for the count. Tracked in #75211.
##########
airflow-core/src/airflow/utils/log/file_task_handler.py:
##########
@@ -647,21 +647,22 @@ def _read(
TaskInstanceState.DEFERRED,
)
- with LogStreamAccumulator(out_stream, HEAP_DUMP_SIZE) as
stream_accumulator:
- log_pos = stream_accumulator.total_lines
- out_stream = stream_accumulator.stream
-
- # skip log stream until the last position
- if metadata and "log_pos" in metadata:
- out_stream = islice(out_stream, metadata["log_pos"], None)
- else:
- # first time reading log, add messages before interleaved log
stream
- out_stream = chain(header, out_stream)
+ out_metadata: LogMetadata = {"end_of_log": end_of_log}
+ if not end_of_log:
+ # Only an unfinished read needs log_pos for the continuation
token, and counting
Review Comment:
Updated both the `:param metadata:` and `:return:` sections in e4d36e30b1d.
--
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]
