1fanwang opened a new pull request, #72312: URL: https://github.com/apache/airflow/pull/72312
An asset that watches an Iceberg table never fires on Airflow 3.3.0 or 3.3.1. The watcher's first poll raises, the triggerer restarts the trigger, and it raises again, so the table sits in a crash loop and no event ever reaches the DAG: ``` AttributeError: 'AssetStateStoreAccessors' object has no attribute 'aget'. Did you mean: 'get'? ``` The watcher keeps a snapshot watermark in the asset state store so that a triggerer restart does not re-emit the current head as a fresh commit, and it awaits the store's `aget`/`aset` to do that. Those async accessors only arrive in https://github.com/apache/airflow/pull/72127. On 3.3.0 and 3.3.1 the store carries the blocking `get`/`set` and nothing else. `main` is broken the same way today, because https://github.com/apache/airflow/pull/72173 merged ahead of that dependency. After this change the watcher polls and emits normally on every version the provider supports, and still keeps its watermark across a triggerer restart. The fix awaits the store's own `aget`/`aset` where it has them and otherwise runs the blocking accessor through `asyncio.to_thread`, which is what the code did before https://github.com/apache/airflow/pull/72173 and what keeps a blocking round trip to the API server off the triggerer's event loop. It duck-types rather than gating on a version constant because https://github.com/apache/airflow/pull/72127 is unmerged, so the release that will carry the async accessors is not yet known. Core picks up `aget_connection` the same way. Also tidies the `:param` lines into full sentences while in the file. related: https://github.com/apache/airflow/pull/72173 related: https://github.com/apache/airflow/pull/72127 ## Testing Done Drove the real `AssetStateStoreAccessors`, built the way `triggerer_job_runner` builds it, through a real `IcebergTableSnapshotTrigger`. The API server is stood in for so the store's blocking accessor has something to answer. <details> <summary>Repro script</summary> ```python import asyncio from contextlib import aclosing, suppress from unittest.mock import MagicMock, patch from airflow.providers.apache.iceberg.triggers.iceberg import IcebergTableSnapshotTrigger from airflow.sdk import Asset from airflow.sdk.execution_time.comms import AssetStateStoreResult from airflow.sdk.execution_time.context import AssetStateStoreAccessors LOAD_TABLE = "airflow.providers.apache.iceberg.hooks.iceberg.IcebergHook.load_table" def table_at(snapshot_id: int) -> MagicMock: table = MagicMock() table.metadata.refs = {"main": MagicMock(snapshot_id=snapshot_id)} return table async def main() -> None: # Built exactly as triggerer_job_runner builds it for a watched asset. store = AssetStateStoreAccessors(inlets=[Asset(name="orders", uri="iceberg://sales.orders")]) print(f"AssetStateStoreAccessors: aget={hasattr(store, 'aget')} aset={hasattr(store, 'aset')}") trigger = IcebergTableSnapshotTrigger(table="sales.orders", poll_interval=0.01) trigger.asset_state_store = store # Stands in for the API server, which answers the store's blocking accessor. comms = MagicMock() comms.send.return_value = AssetStateStoreResult(value=222) payloads = [] with ( patch("airflow.sdk.execution_time.task_runner.SUPERVISOR_COMMS", comms, create=True), patch(LOAD_TABLE, return_value=table_at(222)), ): async with aclosing(trigger.run()) as events: with suppress(asyncio.TimeoutError): payloads.append(await asyncio.wait_for(anext(events), timeout=1.0)) print(f"watermark read: {comms.send.call_args}") print(f"events emitted: {payloads}") assert payloads == [], "the watermark should have suppressed the re-emit of snapshot 222" print("OK: the stored watermark suppressed a re-emit of snapshot 222") asyncio.run(main()) ``` </details> <details> <summary>Raw logs, on <code>main</code> first and then on this branch</summary> ```console $ git checkout upstream/main -- providers/apache/iceberg/src/airflow/providers/apache/iceberg/triggers/iceberg.py $ uv run --project providers/apache/iceberg python repro.py AssetStateStoreAccessors: aget=False aset=False Traceback (most recent call last): File "repro.py", line 64, in <module> asyncio.run(main()) ... File "providers/apache/iceberg/src/airflow/providers/apache/iceberg/triggers/iceberg.py", line 135, in run stored = await store.aget(WATERMARK_KEY) ^^^^^^^^^^ AttributeError: 'AssetStateStoreAccessors' object has no attribute 'aget'. Did you mean: 'get'? $ git checkout HEAD -- providers/apache/iceberg/src/airflow/providers/apache/iceberg/triggers/iceberg.py $ uv run --project providers/apache/iceberg python repro.py AssetStateStoreAccessors: aget=False aset=False watermark read: call(GetAssetStateStoreByName(name='orders', key='snapshot_id', type='GetAssetStateStoreByName')) events emitted: [] OK: the stored watermark suppressed a re-emit of snapshot 222 ``` The two new unit tests fail the same way on `main` and pass here: ```console $ uv run --project providers/apache/iceberg pytest \ providers/apache/iceberg/tests/unit/apache/iceberg/triggers/test_iceberg.py -q 21 passed, 1 warning in 8.73s ``` </details> -- 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]
