Lee-W commented on code in PR #66584: URL: https://github.com/apache/airflow/pull/66584#discussion_r3292224007
########## airflow-core/src/airflow/triggers/shared_stream.py: ########## @@ -0,0 +1,388 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +""" +Shared underlying I/O between :class:`BaseEventTrigger` instances in the triggerer. + +When multiple triggers declare the same non-``None`` +:meth:`~airflow.triggers.base.BaseEventTrigger.shared_stream_key`, the +triggerer routes them through :class:`SharedStreamManager` so that one +underlying poll loop produces raw events that are broadcast to every +participating trigger. Each trigger then runs +:meth:`~airflow.triggers.base.BaseEventTrigger.filter_shared_stream` to +convert the broadcast into its own :class:`~airflow.triggers.base.TriggerEvent` +instances. Triggers that opt out (the default) keep their independent +``run()``-based poll loops untouched. + +Scope and the missing ack channel +--------------------------------- + +The shared-stream channel is **one-way**: events flow from +``open_shared_stream`` out to each subscriber's ``filter_shared_stream``, +with no path back. Subscribers cannot tell the producer "I accepted this +event; please advance / commit / ack". The pattern is therefore only safe +for upstreams whose consumption does not need a producer-side side effect +tied to a subscriber's accept / reject decision: + +* Idempotent / read-only reads (filesystem listings, polling REST APIs). +* Auto-commit Kafka consumers (``enable.auto.commit=true``). +* Subscriber-side-effect cleanup (``unlink``, local marking, …) where the + per-event action goes through APIs the subscriber owns independently. + +Kafka manual-commit consumers, SQS delete-on-process / visibility +extension, and similar message-broker patterns where progress is per-message +and tied to the subscriber's decision are **not** in scope here today. A +producer-side ack channel to cover them is a follow-up that should be +designed against a concrete Kafka or SQS consumer rather than against an +abstract API. See :class:`~airflow.triggers.base.BaseEventTrigger` for the +matching subclass-facing notes. + +Lifecycle invariants +-------------------- + +The manager and groups cooperate to keep a single invariant true at every +``await``-point: + + A key is present in :attr:`SharedStreamManager._groups` only while its + group's poll task is alive and accepting new subscribers. + +This rules out the late-subscriber races that the naive design admits — a +new subscriber for a key whose poll has died or is in the middle of being +torn down always falls through to "create a fresh group" rather than +attaching to a dead one and hanging on an empty queue. The invariant is +maintained synchronously: + +* When ``_poll`` ends for any reason other than cancellation (the upstream + iterator raised, or returned), the group's ``finally`` block evicts the + key from ``_groups`` and broadcasts a terminal sentinel to current + subscribers — all without yielding, so no other coroutine can interleave. +* When the last subscriber leaves, :meth:`SharedStreamManager.unsubscribe` + evicts the key from ``_groups`` *before* awaiting ``group.stop()``, so a + new subscriber arriving while we wait for cancellation creates a fresh + group. +* :meth:`SharedStreamManager.stop_all` clears ``_groups`` in one synchronous + step before awaiting any stop, applying the same rule to shutdown. +""" + +from __future__ import annotations + +import asyncio +from collections.abc import AsyncGenerator, AsyncIterator, Callable, Hashable +from contextlib import suppress +from typing import TYPE_CHECKING, Any + +import structlog + +if TYPE_CHECKING: + from structlog.stdlib import BoundLogger + + from airflow.triggers.base import BaseEventTrigger + +log = structlog.get_logger(__name__) + +DEFAULT_SUBSCRIBER_QUEUE_MAX = 1024 +"""Default per-subscriber queue size for shared streams. + +The :class:`SharedStreamManager` admits up to this many unconsumed raw events +per subscriber before treating the subscriber as too slow to keep up — at +which point the subscriber's trigger is failed with +:class:`_SubscriberOverflow` rather than the queue growing without bound. + +Used as the fallback when no value is passed to ``SharedStreamManager``; +in the triggerer this is overridden from the +``[triggerer] shared_stream_subscriber_queue_size`` config option. +""" + + +class _PollTerminated(Exception): Review Comment: `_PollTerminated` and `_SubscriberOverflow` are internal sentinels — they never escape `SharedStreamManager` and the trigger-failure path catches `Exception` regardless of base. I'd rather not add more `AirflowException` subclasses for purely internal types. -- 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]
