Nik-Tenzo commented on PR #73004:
URL: https://github.com/apache/airflow/pull/73004#issuecomment-6577601794
This isn't only a Triggerer problem: on Airflow 3.3 the Celery executor runs
each task's supervisor inside the long-lived pool process, so `upload()` runs
once per task try on the same `CloudWatchRemoteLogIO`, and every try leaves a
watchtower thread, queue and sequence token behind.
On Airflow 3.3.2, amazon provider 9.21.0, watchtower 3.4.0 and Python 3.14,
each pool process ran at 2 threads plus one per task it had run (52 at 50
tasks, 102 at 100), and RSS grew ~0.22 MiB per task. Our 8 GB workers at
concurrency 20 ran out of memory before `worker_max_tasks_per_child=1000`
recycled the processes. Running `main`'s `CloudWatchRemoteLogIO` locally
against a fake client gave the same pattern: 1,001 threads after 1,000 tasks,
and `upload()` (via `flush()` over every queue) took 74 ms by then.
As a stopgap we end the finished stream in a subclass's `upload()`. That
keeps pool processes at 2–3 threads with flat memory at 1,000 tasks per process:
```python
def upload(self, path, ti=None):
super().upload(path, ti)
stream_name = Path(path).as_posix().replace(":", "_")
handler = self.handler
stream_queue = handler.queues.pop(stream_name, None)
if stream_queue is not None:
stream_queue.put(handler.END)
with stream_queue.all_tasks_done:
ended = stream_queue.all_tasks_done.wait_for(
lambda: stream_queue.unfinished_tasks == 0,
timeout=handler.FLUSH_TIMEOUT
)
if ended:
handler.sequence_tokens.pop(stream_name, None)
handler.threads = [thread for thread in handler.threads if
thread.is_alive()]
```
This PR's per-stream handlers do the same without touching watchtower's
internals, so +1 to it. It might be worth mentioning workers in the title or
description, since that's the more common place to hit it.
---
Drafted-by: Claude Code (Claude Opus 5.5); reviewed by @Nik-Tenzo before
posting
--
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]
