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 1f44dac375cb8..15877d3d9d0b2 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)]