You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "fpiped (via GitHub)" <gi...@apache.org> on 2026/09/15 17:04:56 UTC
[PR] Read the AWS task logs once more after the fetcher is stopped [airflow]
fpiped opened a new pull request, #73211:
URL: https://github.com/apache/airflow/pull/73211
`AwsTaskLogFetcher.run` checks the stop flag at the top of its loop, then sleeps, then fetches. When `stop()` arrives while the thread is inside a fetch, or in the gap between a fetch finishing and the next check, the loop ends with no read after the stop, and the events the container wrote at the end of the task are never forwarded to the task log.
Both callers stop the fetcher as soon as the task or job has ended, which is exactly when those events appear: `EcsRunTaskOperator.execute` (`finally: self.task_log_fetcher.stop()`) and `BatchClientHook.wait_for_job` (`finally: batch_log_fetcher.stop()`).
This forwards the events once more after leaving the loop. The continuation token keeps the extra read from repeating events already seen, and `AwsLogsHook.get_log_events` returns as soon as the stream is exhausted, so the cost is one `get_log_events` call at thread exit. The body of the loop moved into `_forward_log_events` unchanged.
`test_run_forwards_the_events_written_before_it_was_stopped` covers it: it fails on the current code (one fetch, the last event never logged) and passes here. The two existing `run` tests each get one more empty page in their `side_effect`, since `run` now always performs that final fetch.
Verified locally with `apache-airflow 3.1.8` and the provider installed from this branch:
```
providers/amazon/tests/unit/amazon/aws/utils/test_task_log_fetcher.py 32 passed
ruff check / ruff format --check on both files clean
```
`tests/unit/amazon/aws/operators/test_ecs.py`, `test_batch.py` and `hooks/test_batch_client.py` patch `AwsTaskLogFetcher` as a whole and never exercise `run`, so they are unaffected; CI runs them.
closes: #73210
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes (please specify the tool below)
Generated-by: Claude Code following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
The behaviour was reproduced against the released provider before the change, the fix and its test were reviewed line by line, and the tests and static checks above were run locally.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Read the AWS task logs once more after the fetcher is stopped [airflow]
Posted by "o-nikolas (via GitHub)" <gi...@apache.org>.
o-nikolas commented on code in PR #73211:
URL: https://github.com/apache/airflow/pull/73211#discussion_r4095954279
##########
providers/amazon/src/airflow/providers/amazon/aws/utils/task_log_fetcher.py:
##########
@@ -102,22 +102,25 @@ def run(self) -> None:
continuation_token = AwsLogsHook.ContinuationToken()
while not self.is_stopped():
time.sleep(self.fetch_interval.total_seconds())
Review Comment:
You could use `self._event.wait(self.fetch_interval.total_seconds())` to stop sooner (when the stop event is received) rather than the static wait time. That would allow an earlier exist in some cases and save a bit of time. But not critical.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Read the AWS task logs once more after the fetcher is stopped [airflow]
Posted by "o-nikolas (via GitHub)" <gi...@apache.org>.
o-nikolas merged PR #73211:
URL: https://github.com/apache/airflow/pull/73211
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Read the AWS task logs once more after the fetcher is stopped [airflow]
Posted by "fpiped (via GitHub)" <gi...@apache.org>.
fpiped commented on code in PR #73211:
URL: https://github.com/apache/airflow/pull/73211#discussion_r4106392768
##########
providers/amazon/src/airflow/providers/amazon/aws/utils/task_log_fetcher.py:
##########
@@ -102,22 +102,25 @@ def run(self) -> None:
continuation_token = AwsLogsHook.ContinuationToken()
while not self.is_stopped():
time.sleep(self.fetch_interval.total_seconds())
Review Comment:
Done in e3998c6c8d, with a test that `stop()` ends the wait. Thanks!
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Read the AWS task logs once more after the fetcher is stopped [airflow]
Posted by "boring-cyborg[bot] (via GitHub)" <gi...@apache.org>.
boring-cyborg[bot] commented on PR #73211:
URL: https://github.com/apache/airflow/pull/73211#issuecomment-5879704278
Awesome work, congrats on your first merged pull request! You are invited to check our [Issue Tracker](https://github.com/apache/airflow/issues) for additional contributions.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org