(airflow) branch main updated: Fix flaky Iceberg snapshot trigger tests under CI thread-pool latency (#71825)

[weilee]

This is an automated email from the ASF dual-hosted git repository.

Lee-W pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git


The following commit(s) were added to refs/heads/main by this push:
     new 7f524b46434 Fix flaky Iceberg snapshot trigger tests under CI 
thread-pool latency (#71825)
7f524b46434 is described below

commit 7f524b464347d32dfeed2243babf7a0c6dff8f2a
Author: Wei Lee <wei...@apache.org>
AuthorDate: Thu Aug 20 10:38:21 2026 +0800

    Fix flaky Iceberg snapshot trigger tests under CI thread-pool latency 
(#71825)
---
 .../iceberg/tests/unit/apache/iceberg/triggers/test_iceberg.py    | 8 ++++++--
 1 file changed, 6 insertions(+), 2 deletions(-)

diff --git 
a/providers/apache/iceberg/tests/unit/apache/iceberg/triggers/test_iceberg.py 
b/providers/apache/iceberg/tests/unit/apache/iceberg/triggers/test_iceberg.py
index 1f44dac375c..15877d3d9d0 100644
--- 
a/providers/apache/iceberg/tests/unit/apache/iceberg/triggers/test_iceberg.py
+++ 
b/providers/apache/iceberg/tests/unit/apache/iceberg/triggers/test_iceberg.py
@@ -135,7 +135,9 @@ async def test_emits_once_per_new_snapshot():
     """Each commit produces exactly one event carrying the snapshot it 
replaced."""
     with patch(LOAD_TABLE, side_effect=[_table_at(111), _table_at(222), 
_table_at(222), _table_at(333)]):
         trigger = IcebergTableSnapshotTrigger(table="db.tbl", 
poll_interval=0.01, last_seen_snapshot_id=111)
-        payloads = await _collect(trigger, 2)
+        # Gathering 2 events takes 4 polling rounds, each with a real 
asyncio.to_thread call;
+        # the default 1s budget is too tight under CI thread-pool scheduling 
latency.
+        payloads = await _collect(trigger, 2, timeout=3.0)
 
     assert [(p["previous_snapshot_id"], p["snapshot_id"]) for p in payloads] 
== [(111, 222), (222, 333)]
 
@@ -190,7 +192,9 @@ async def test_persists_the_watermark_on_each_event():
     trigger.asset_state_store = store
 
     with patch(LOAD_TABLE, side_effect=[_table_at(111), _table_at(222), 
_table_at(222)]):
-        payloads = await _collect(trigger, 2)
+        # Gathering 2 events runs several real asyncio.to_thread calls (head 
lookup + store
+        # get/set); the default 1s budget is too tight under CI thread-pool 
scheduling latency.
+        payloads = await _collect(trigger, 2, timeout=3.0)
 
     assert [p["snapshot_id"] for p in payloads] == [111, 222]
     assert [c.args for c in store.set.call_args_list] == [("snapshot_id", 
111), ("snapshot_id", 222)]