This is an automated email from the ASF dual-hosted git repository.
vincbeck 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 39c6e268811 Fix Iceberg watcher crash by reverting to the sync state
store (#72312)
39c6e268811 is described below
commit 39c6e268811583ec6d797c52732a29a5660c07da
Author: Stefan Wang <[email protected]>
AuthorDate: Wed Sep 9 10:43:24 2026 -0700
Fix Iceberg watcher crash by reverting to the sync state store (#72312)
The async accessors adopted in
https://github.com/apache/airflow/pull/72173 exist in no released Airflow,
so
the watcher raises AttributeError on its first poll. The blocking accessors
work on every supported version. The added test builds a real
AssetStateStoreAccessors, which a mock cannot stand in for.
---
.../providers/apache/iceberg/triggers/iceberg.py | 4 +-
.../unit/apache/iceberg/triggers/test_iceberg.py | 51 +++++++++++++++++-----
2 files changed, 42 insertions(+), 13 deletions(-)
diff --git
a/providers/apache/iceberg/src/airflow/providers/apache/iceberg/triggers/iceberg.py
b/providers/apache/iceberg/src/airflow/providers/apache/iceberg/triggers/iceberg.py
index 89cecfd5d09..63f8d28ea49 100644
---
a/providers/apache/iceberg/src/airflow/providers/apache/iceberg/triggers/iceberg.py
+++
b/providers/apache/iceberg/src/airflow/providers/apache/iceberg/triggers/iceberg.py
@@ -125,7 +125,7 @@ class IcebergTableSnapshotTrigger(BaseEventTrigger):
store = getattr(self, "asset_state_store", None)
if store is not None:
try:
- stored = await store.aget(WATERMARK_KEY)
+ stored = await asyncio.to_thread(store.get, WATERMARK_KEY)
except ValueError as err:
# The accessor serves one asset at a time, so it refuses to
guess when this
# trigger is watched by several. That happens because triggers
are deduplicated
@@ -151,7 +151,7 @@ class IcebergTableSnapshotTrigger(BaseEventTrigger):
if head is not None and head != self.last_seen_snapshot_id:
previous, self.last_seen_snapshot_id =
self.last_seen_snapshot_id, head
if store is not None:
- await store.aset(WATERMARK_KEY, head)
+ await asyncio.to_thread(store.set, WATERMARK_KEY, head)
yield TriggerEvent(
{
"table": self.table,
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 2099d2767ed..a1127448957 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
@@ -19,13 +19,15 @@ from __future__ import annotations
import asyncio
from contextlib import aclosing, suppress
from typing import TYPE_CHECKING
-from unittest.mock import AsyncMock, MagicMock, patch
+from unittest.mock import MagicMock, patch
import pytest
from pyiceberg.exceptions import NoSuchNamespaceError, NoSuchTableError
from airflow.providers.apache.iceberg.triggers.iceberg import
IcebergTableSnapshotTrigger
+from tests_common.test_utils.version_compat import AIRFLOW_V_3_3_PLUS
+
if TYPE_CHECKING:
from collections.abc import AsyncGenerator
@@ -171,7 +173,7 @@ async def test_resumes_from_the_stored_watermark():
row was written; only the stored watermark reflects what was actually
emitted.
"""
store = MagicMock()
- store.aget = AsyncMock(return_value=222)
+ store.get.return_value = 222
trigger = IcebergTableSnapshotTrigger(table="db.tbl", poll_interval=0.01,
last_seen_snapshot_id=111)
trigger.asset_state_store = store
@@ -180,32 +182,31 @@ async def test_resumes_from_the_stored_watermark():
payloads = await _collect(trigger, 1, timeout=0.2)
assert payloads == []
- store.aget.assert_awaited_once_with("snapshot_id")
+ store.get.assert_called_once_with("snapshot_id")
@pytest.mark.asyncio
async def test_persists_the_watermark_on_each_event():
store = MagicMock()
- store.aget = AsyncMock(return_value=None)
- store.aset = AsyncMock()
+ store.get.return_value = None
trigger = IcebergTableSnapshotTrigger(table="db.tbl", poll_interval=0.01)
trigger.asset_state_store = store
with patch(LOAD_TABLE, side_effect=[_table_at(111), _table_at(222),
_table_at(222)]):
- # Multiple real thread-pool head lookups can exceed the default 1s
budget under CI latency.
+ # 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.aset.await_args_list] == [("snapshot_id",
111), ("snapshot_id", 222)]
+ assert [c.args for c in store.set.call_args_list] == [("snapshot_id",
111), ("snapshot_id", 222)]
@pytest.mark.asyncio
async def test_runs_without_a_watermark_when_several_assets_watch_it():
"""More than one watched asset leaves no single cursor, so it degrades
instead of raising."""
store = MagicMock()
- store.aget = AsyncMock(side_effect=ValueError("Task has 2 concrete inlets
and outlets"))
- store.aset = AsyncMock()
+ store.get.side_effect = ValueError("Task has 2 concrete inlets and
outlets")
trigger = IcebergTableSnapshotTrigger(table="db.tbl", poll_interval=0.01)
trigger.asset_state_store = store
@@ -214,14 +215,14 @@ async def
test_runs_without_a_watermark_when_several_assets_watch_it():
payloads = await _collect(trigger, 1)
assert [p["snapshot_id"] for p in payloads] == [111]
- store.aset.assert_not_awaited()
+ store.set.assert_not_called()
@pytest.mark.asyncio
async def test_a_state_store_failure_is_not_mistaken_for_several_assets():
"""A pluggable backend can raise ValueError too, and hiding it would
disable the watermark."""
store = MagicMock()
- store.aget = AsyncMock(side_effect=ValueError("could not decode the stored
reference"))
+ store.get.side_effect = ValueError("could not decode the stored reference")
trigger = IcebergTableSnapshotTrigger(table="db.tbl", poll_interval=0.01)
trigger.asset_state_store = store
@@ -231,6 +232,34 @@ async def
test_a_state_store_failure_is_not_mistaken_for_several_assets():
await _collect(trigger, 1)
[email protected]
[email protected](not AIRFLOW_V_3_3_PLUS, reason="asset_state_store arrived
in Airflow 3.3.0")
+async def test_reads_the_watermark_through_the_real_asset_state_store():
+ """A mocked store invents whatever method the trigger reaches for, so it
cannot show that
+ the accessor the triggerer really injects offers that method.
+ """
+ from airflow.sdk import Asset
+ from airflow.sdk.execution_time.comms import AssetStateStoreResult
+ from airflow.sdk.execution_time.context import AssetStateStoreAccessors
+
+ # Built the way triggerer_job_runner builds it for a watched asset.
+ store = AssetStateStoreAccessors(inlets=[Asset(name="orders",
uri="iceberg://db.tbl")])
+ comms = MagicMock()
+ comms.send.return_value = AssetStateStoreResult(value=222)
+
+ trigger = IcebergTableSnapshotTrigger(table="db.tbl", poll_interval=0.01,
last_seen_snapshot_id=111)
+ trigger.asset_state_store = store
+
+ with (
+ patch("airflow.sdk.execution_time.task_runner.SUPERVISOR_COMMS",
comms, create=True),
+ patch(LOAD_TABLE, return_value=_table_at(222)),
+ ):
+ payloads = await _collect(trigger, 1, timeout=0.5)
+
+ assert payloads == []
+ assert comms.send.call_args.args[0].key == "snapshot_id"
+
+
@pytest.mark.asyncio
async def test_runs_on_airflow_without_an_asset_state_store():
"""``asset_state_store`` postdates the oldest Airflow this provider
supports."""
