Flaky shared_stream test times out in CI

[AlejandroMorgante]

airflow-core/tests/unit/triggers/test_shared_stream.py::test_ack_mode_queue_full_during_fanout_does_not_break_iteration is timing out in CI.

I hit this in PR #68479, but it does not appear related to that PR. The same test is also failing on main with the same TimeoutError.

Main failure example:
https://github.com/apache/airflow/actions/runs/27607202688/job/81625031694

Relevant failure:

airflow-core/tests/unit/triggers/test_shared_stream.py:1437: in test_ack_mode_queue_full_during_fanout_does_not_break_iteration
    await asyncio.wait_for(collect_full(), timeout=2.0)
...
E   TimeoutError

This seems to be a flaky or recently introduced timing issue in the shared stream trigger tests and is currently blocking unrelated PRs when the full core test suite runs.

[shahar1]

@Lee-W could be related to #67523?

[venkatamandavilli-code]

Thanks for reporting this.

This looks like a flaky timing-related failure rather than something specific to the linked PR, especially since the same test is also failing on main with a TimeoutError.

Since the timeout occurs in test_ack_mode_queue_full_during_fanout_does_not_break_iteration, it may be worth checking whether the test is relying on timing assumptions that are too tight for CI environments. The asyncio.wait_for(..., timeout=2.0) timeout may not be sufficient under slower or loaded CI runners, especially if queue backpressure or fanout scheduling is involved.

It may also be useful to confirm whether the collector coroutine can get blocked waiting for an item that is never yielded after the queue-full condition is triggered.

This seems important to stabilize because it can block unrelated PRs when the full core test suite runs.

[Lee-W]

I think #68651 makes sense. Let's see how it goes

[Lee-W]

cc @FrankYang0529