d-hervas commented on code in PR #73004:
URL: https://github.com/apache/airflow/pull/73004#discussion_r4000314047
##########
providers/amazon/src/airflow/providers/amazon/aws/log/cloudwatch_task_handler.py:
##########
@@ -168,32 +176,70 @@ def handler(self) -> watchtower.CloudWatchLogHandler:
Rebuild only while the IO is live: once :meth:`close` has run, keep
the closed handler
so a late record is dropped instead of spawning an orphan handler and
background thread.
"""
- if self._cached_handler is None or (not self._closed and
self._cached_handler.shutting_down):
- self._cached_handler = self._build_handler()
- return self._cached_handler
+ with self._stream_lock:
+ if self._cached_handler is None or (not self._closed and
self._cached_handler.shutting_down):
+ self._cached_handler = self._build_handler()
+ return self._cached_handler
+
+ def _get_stream_handler(self, stream_name: str) ->
watchtower.CloudWatchLogHandler | None:
+ """Return the live handler for ``stream_name`` while holding
``_stream_lock``."""
+ if self._closed or self._building_stream_handler or stream_name in
self._closing_streams:
+ return None
+
+ handler = self._stream_handlers.get(stream_name)
+ if handler is not None and not handler.shutting_down:
+ return handler
+
+ self._stream_handlers.pop(stream_name, None)
+ if (
+ not self._stream_handlers
+ and self._cached_handler is not None
+ and not self._cached_handler.shutting_down
+ ):
+ handler = self._cached_handler
+ handler.log_stream_name = stream_name
+ else:
+ self._building_stream_handler = True
+ try:
+ handler = self._build_handler(stream_name)
+ finally:
+ self._building_stream_handler = False
+ self._stream_handlers[stream_name] = handler
+ self._cached_handler = handler
+ return handler
+
+ def _close_stream(self, stream_name: str) -> None:
+ with self._stream_lock:
+ handler = self._stream_handlers.pop(stream_name, None)
+ if handler is None:
+ return
+ self._closing_streams.add(stream_name)
+ if self._cached_handler is handler:
+ self._cached_handler =
next(reversed(self._stream_handlers.values()), None)
+
+ try:
+ handler.close()
Review Comment:
`close()` is synchronous, but so was the previous Watchtower `flush()`.
Watchtower's `flush()` signals every queue owned by the handler and waits for
them to drain. Because the shared handler retained queues for completed
streams,
each upload could revisit all accumulated queues.
The new handler owns one stream, so `close()` waits only for that completed
stream's queue before terminating its worker. The call is made by the
synchronous `TriggerRunnerSupervisor` request handler rather than directly
in the
trigger coroutine event loop. A slow CloudWatch request can delay the
supervisor, but that could already happen with `flush()`. This binds the
wait to just one stream instead of all retained streams.
--
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]
