hussein-awala opened a new pull request, #76217: URL: https://github.com/apache/airflow/pull/76217
On every state sync that reports persisted trigger events, the triggerer passes their sequence numbers to `SharedStreamManager.confirm_persisted` ([call site](https://github.com/apache/airflow/blob/b95498cd56/airflow-core/src/airflow/jobs/triggerer_job_runner.py#L1689-L1690)). The manager offers the whole batch to every live group, fast-path watcher groups included, although exactly one group owns each number ([broadcast](https://github.com/apache/airflow/blob/b95498cd56/airflow-core/src/airflow/triggers/shared_stream.py#L1261-L1273)). Syncs run about once per second on the event loop that runs every trigger coroutine, so a triggerer with many groups next to a busy ack-mode stream pays groups × events work at each such sync. Route each sequence number to the group that drew it and keep that routing state bounded. related: #76209 The manager records the owning group's key in `bind_pending_event`. Each group reports every sequence number that leaves its pending-confirmation index, so an entry lives until its sequence number is confirmed, its binding resolves, or its group's poll ends. A confirmation for a group that has already left the manager is ignored; if a newer group has meanwhile taken the same key, it ignores those numbers too, since they are unique per manager. In both cases the old group's poll teardown removes the entries and the broker redelivers the gated events, as before. Event ordering, advance batching and resolution outcomes are unchanged, with no new configuration or schema change. `SharedStreamManager` keeps its signatures; the private `_SharedStreamGroup` gains a required `on_seq_resolved` callback. Only ack-mode shared streams (triggers that override `create_shared_stream_producer`) are affected; fast-path streams draw no sequence numbers and write nothing to the new map. In a local micro-benchmark against the real `SharedStreamManager` with 1,000 live ack-mode groups: | Scenario | Before | After | |---|---:|---:| | `confirm_persisted`, 100 sequence numbers bound in one group | 2305.6 µs | 29.9 µs | | `confirm_persisted` for already-resolved sequence numbers | 2283.1 µs | 2.3 µs | Each variant ran three times, alternating base and head with the order rotated each round; a run reports the median per-call time over 7 repeats of 100 calls with garbage collection off, and the table shows the median of three runs (first row: base 2287.4–2330.2 µs, head 29.6–30.9 µs). No unchanged control was measured. With 50 groups holding 20 never-confirmed sequence numbers each, the new map was empty after ack-timeout resolution and after the groups stopped. These are local measurements of the manager call, not end-to-end triggerer latency. `test_confirm_persisted_routes_seqs_to_owning_group_only` spies on the group's `confirm_persisted` and fails if the manager broadcasts again. Routing and lifecycle tests also assert the map empties at each eviction point. Validation: - Triggerer job, trigger-model and shared-stream tests: 289 passed, 1 skipped on SQLite. - PostgreSQL and MySQL were not run locally; this change has no database code. - Regular and manual static checks: re-running on the final revision. - Selected core suites and the documentation build were not run locally; CI validation is pending. --- ##### Was generative AI tooling used to co-author this PR? - [X] Yes — Claude Code (Opus 5.5) Generated-by: Claude Code (Opus 5.5) following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions) 🤖 Generated with [Claude Code](https://claude.com/claude-code) -- 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]
