This is an automated email from the ASF dual-hosted git repository.
jason810496 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 33e64a1dd67 Follow-up fixes for double-encode trigger PR (#64823)
33e64a1dd67 is described below
commit 33e64a1dd679b9b6df4d04872c1781929c976102
Author: Jason(Zhe-You) Liu <[email protected]>
AuthorDate: Tue Apr 21 14:13:08 2026 +0800
Follow-up fixes for double-encode trigger PR (#64823)
* Follow-up fixes for double-encode trigger PR
* Addressed review comments
---
airflow-core/src/airflow/models/trigger.py | 4 +-
.../tests/unit/dag_processing/test_collection.py | 10 ++--
.../tests/unit/serialization/test_encoders.py | 59 +++++++++++++++++-----
3 files changed, 51 insertions(+), 22 deletions(-)
diff --git a/airflow-core/src/airflow/models/trigger.py
b/airflow-core/src/airflow/models/trigger.py
index d17af8532e0..97079952885 100644
--- a/airflow-core/src/airflow/models/trigger.py
+++ b/airflow-core/src/airflow/models/trigger.py
@@ -209,7 +209,7 @@ class Trigger(Base):
@provide_session
def fetch_trigger_ids_with_non_task_associations(cls, session: Session =
NEW_SESSION) -> set[str]:
"""Fetch all trigger IDs actively associated with non-task entities
like assets and callbacks."""
- from airflow.models.callback import Callback
+ from airflow.models.callback import Callback # to avoid circular
import: Callback -> Trigger
query = select(AssetWatcherModel.trigger_id).union_all(
select(Callback.trigger_id).where(Callback.trigger_id.is_not(None))
@@ -409,7 +409,7 @@ class Trigger(Base):
:param queues: The optional set of trigger queues to filter triggers
by.
:param session: The database session.
"""
- from airflow.models.callback import Callback
+ from airflow.models.callback import Callback # to avoid circular
import: Callback -> Trigger
result: list[Row[Any]] = []
diff --git a/airflow-core/tests/unit/dag_processing/test_collection.py
b/airflow-core/tests/unit/dag_processing/test_collection.py
index 1792c9b38c0..8dd02fd5a88 100644
--- a/airflow-core/tests/unit/dag_processing/test_collection.py
+++ b/airflow-core/tests/unit/dag_processing/test_collection.py
@@ -53,12 +53,14 @@ from airflow.models.dag import DagTag
from airflow.models.dagbundle import DagBundleModel
from airflow.models.errors import ParseImportError
from airflow.models.serialized_dag import SerializedDagModel
+from airflow.models.trigger import Trigger
from airflow.providers.standard.operators.empty import EmptyOperator
from airflow.providers.standard.triggers.file import FileDeleteTrigger
from airflow.sdk import DAG, Asset, AssetAlias, AssetWatcher
from airflow.serialization.definitions.assets import SerializedAsset
-from airflow.serialization.encoders import ensure_serialized_asset
+from airflow.serialization.encoders import encode_trigger,
ensure_serialized_asset
from airflow.serialization.serialized_objects import LazyDeserializedDAG
+from airflow.triggers.base import BaseEventTrigger
from airflow.utils.types import DagRunType
from tests_common.test_utils.config import conf_vars
@@ -188,10 +190,6 @@ class TestAssetModelOperation:
from the DB-stored Trigger row. A mismatch causes the scheduler to
recreate trigger rows on every heartbeat.
"""
- from airflow.models.trigger import Trigger
- from airflow.serialization.encoders import encode_trigger
- from airflow.triggers.base import BaseEventTrigger
-
trigger = FileDeleteTrigger(filepath="/tmp/test.txt",
poke_interval=5.0)
asset = Asset(
"test_hash_consistency_asset",
@@ -236,8 +234,6 @@ class TestAssetModelOperation:
"""Calling add_asset_trigger_references twice with the same trigger
must not create duplicate rows.
"""
- from airflow.models.trigger import Trigger
-
trigger = FileDeleteTrigger(filepath="/tmp/test.txt",
poke_interval=5.0)
asset = Asset(
"test_idempotent_asset",
diff --git a/airflow-core/tests/unit/serialization/test_encoders.py
b/airflow-core/tests/unit/serialization/test_encoders.py
index 479af95ae88..1348e8c6b76 100644
--- a/airflow-core/tests/unit/serialization/test_encoders.py
+++ b/airflow-core/tests/unit/serialization/test_encoders.py
@@ -16,6 +16,9 @@
# under the License.
from __future__ import annotations
+from typing import Any
+from unittest.mock import AsyncMock
+
import pytest
from sqlalchemy import delete
@@ -25,29 +28,62 @@ from airflow.serialization.encoders import encode_trigger
from airflow.serialization.enums import DagAttributeTypes as DAT, Encoding
from airflow.triggers.base import BaseEventTrigger
-pytest.importorskip("airflow.providers.apache.kafka")
-from airflow.providers.apache.kafka.triggers.await_message import
AwaitMessageTrigger
+
+class _CallableKwargsTrigger(BaseEventTrigger):
+ """Mock trigger whose kwargs include non-primitive types (tuples, dicts,
lists).
+
+ This exercises the same serialization edge case as real provider triggers
+ (e.g. Kafka's AwaitMessageTrigger) that pass callable-style kwargs, without
+ requiring the provider to be installed.
+ """
+
+ def __init__(
+ self,
+ topics: tuple[str, ...] | list[str] = (),
+ apply_function: str | None = None,
+ apply_function_args: list[Any] | None = None,
+ apply_function_kwargs: dict[str, Any] | None = None,
+ poll_timeout: float = 1,
+ ) -> None:
+ self.topics = topics
+ self.apply_function = apply_function
+ self.apply_function_args = apply_function_args or ()
+ self.apply_function_kwargs = apply_function_kwargs or {}
+ self.poll_timeout = poll_timeout
+
+ def serialize(self) -> tuple[str, dict[str, Any]]:
+ return (
+ f"{self.__class__.__module__}.{self.__class__.__qualname__}",
+ {
+ "topics": self.topics,
+ "apply_function": self.apply_function,
+ "apply_function_args": self.apply_function_args,
+ "apply_function_kwargs": self.apply_function_kwargs,
+ "poll_timeout": self.poll_timeout,
+ },
+ )
+
+ run = AsyncMock()
+
# Trigger fixtures covering primitive-only kwargs (FileDeleteTrigger) and
-# non-primitive kwargs like tuple/dict (AwaitMessageTrigger).
+# non-primitive kwargs like tuple/dict (_CallableKwargsTrigger).
_TRIGGER_PARAMS = [
pytest.param(
FileDeleteTrigger(filepath="/tmp/test.txt", poke_interval=5.0),
id="primitive_kwargs_only",
),
- pytest.param(AwaitMessageTrigger(topics=()), id="empty_tuple"),
+ pytest.param(_CallableKwargsTrigger(topics=()), id="empty_tuple"),
pytest.param(
- AwaitMessageTrigger(topics=("fizz_buzz",), poll_timeout=1.0,
commit_offset=True),
+ _CallableKwargsTrigger(topics=("fizz_buzz",), poll_timeout=1.0),
id="single_topic_tuple",
),
pytest.param(
- AwaitMessageTrigger(
+ _CallableKwargsTrigger(
topics=["t1", "t2"],
apply_function="my.module.func",
apply_function_args=["a", "b"],
apply_function_kwargs={"key": "value"},
- kafka_config_id="my_kafka",
- poll_interval=2,
poll_timeout=3,
),
id="all_non_primitive_kwargs",
@@ -67,17 +103,14 @@ class TestEncodeTrigger:
def test_encode_from_trigger_object(self):
"""Non-primitive kwargs are properly serialized from a trigger
object."""
- trigger = AwaitMessageTrigger(topics=())
+ trigger = _CallableKwargsTrigger(topics=())
result = encode_trigger(trigger)
- assert (
- result["classpath"] ==
"airflow.providers.apache.kafka.triggers.await_message.AwaitMessageTrigger"
- )
+ assert result["classpath"].endswith("_CallableKwargsTrigger")
# tuple kwarg is wrapped by BaseSerialization
assert result["kwargs"]["topics"] == {Encoding.TYPE: DAT.TUPLE,
Encoding.VAR: []}
# Primitives pass through as-is
assert result["kwargs"]["poll_timeout"] == 1
- assert result["kwargs"]["commit_offset"] is True
def test_encode_file_delete_trigger(self):
"""Primitive-only kwargs pass through without wrapping."""
