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 0652ebe94e0 Fix deferred dbt Cloud tasks failing with tenacity
retry_args (#74489)
0652ebe94e0 is described below
commit 0652ebe94e0216640be09f988d52e60cb9f5620a
Author: PoAn Yang <[email protected]>
AuthorDate: Wed Oct 14 16:10:40 2026 +0800
Fix deferred dbt Cloud tasks failing with tenacity retry_args (#74489)
Signed-off-by: PoAn Yang <[email protected]>
---
.../airflow/providers/dbt/cloud/triggers/dbt.py | 24 ++++++++++++++++++++--
.../tests/unit/dbt/cloud/triggers/test_dbt.py | 20 ++++++++++++++++++
2 files changed, 42 insertions(+), 2 deletions(-)
diff --git
a/providers/dbt/cloud/src/airflow/providers/dbt/cloud/triggers/dbt.py
b/providers/dbt/cloud/src/airflow/providers/dbt/cloud/triggers/dbt.py
index f2525406221..9a66c58be68 100644
--- a/providers/dbt/cloud/src/airflow/providers/dbt/cloud/triggers/dbt.py
+++ b/providers/dbt/cloud/src/airflow/providers/dbt/cloud/triggers/dbt.py
@@ -17,6 +17,7 @@
from __future__ import annotations
import asyncio
+import json
import time
from collections.abc import AsyncIterator
from typing import Any
@@ -38,7 +39,11 @@ class DbtCloudRunJobTrigger(BaseTrigger):
the task is considered timed out.
:param account_id: The ID of a dbt Cloud account.
:param poll_interval: polling period in seconds to check for the status.
- :param hook_params: Extra arguments passed to the DbtCloudHook constructor.
+ :param hook_params: Extra arguments passed to the DbtCloudHook
constructor. Airflow serializes
+ trigger arguments to store them in the metadata database, so the
trigger keeps only the
+ arguments whose values can be serialized to JSON. For example, the
trigger drops a
+ ``retry_args`` value that holds tenacity objects, and the hook in the
trigger then retries
+ according to ``retry_limit`` and ``retry_delay``.
"""
def __init__(
@@ -58,7 +63,22 @@ class DbtCloudRunJobTrigger(BaseTrigger):
self.end_time = end_time
self.execution_deadline = execution_deadline
self.poll_interval = poll_interval
- self.hook_params = hook_params or {}
+ # Since Airflow 3.2, the task runner serializes trigger arguments with
airflow.sdk.serde instead
+ # of BaseSerialization. When the task defers, serde raises TypeError
for a type that it has no
+ # serializer for, such as a tenacity object. The operator has already
started the dbt Cloud job
+ # before the task defers. Before 3.2, BaseSerialization stores the
same value as a string.
+ self.hook_params: dict[str, Any] = {}
+ for name, value in (hook_params or {}).items():
+ try:
+ json.dumps(value)
+ except (TypeError, ValueError):
+ self.log.warning(
+ "The trigger drops hook_params[%r] because the value
cannot be serialized to JSON. "
+ "The DbtCloudHook in the trigger uses the default value
for this argument instead.",
+ name,
+ )
+ else:
+ self.hook_params[name] = value
def serialize(self) -> tuple[str, dict[str, Any]]:
"""Serialize DbtCloudRunJobTrigger arguments and classpath."""
diff --git a/providers/dbt/cloud/tests/unit/dbt/cloud/triggers/test_dbt.py
b/providers/dbt/cloud/tests/unit/dbt/cloud/triggers/test_dbt.py
index 63376e562e6..fb309726c29 100644
--- a/providers/dbt/cloud/tests/unit/dbt/cloud/triggers/test_dbt.py
+++ b/providers/dbt/cloud/tests/unit/dbt/cloud/triggers/test_dbt.py
@@ -23,6 +23,7 @@ from unittest import mock
from unittest.mock import AsyncMock
import pytest
+from tenacity import stop_after_attempt, wait_fixed
from airflow.providers.dbt.cloud.hooks.dbt import DbtCloudHook,
DbtCloudJobRunStatus
from airflow.providers.dbt.cloud.triggers.dbt import DbtCloudRunJobTrigger
@@ -68,6 +69,25 @@ class TestDbtCloudRunJobTrigger:
"hook_params": {"retry_delay": 10},
}
+ def test_serialization_drops_hook_params_that_cannot_be_serialized(self,
end_time, caplog):
+ trigger = DbtCloudRunJobTrigger(
+ conn_id=self.CONN_ID,
+ poll_interval=self.POLL_INTERVAL,
+ end_time=end_time,
+ run_id=self.RUN_ID,
+ account_id=self.ACCOUNT_ID,
+ hook_params={
+ "retry_limit": 3,
+ "retry_delay": 2.0,
+ "retry_args": {"stop": stop_after_attempt(3), "wait":
wait_fixed(1)},
+ },
+ )
+
+ _, kwargs = trigger.serialize()
+
+ assert kwargs["hook_params"] == {"retry_limit": 3, "retry_delay": 2.0}
+ assert any("hook_params['retry_args']" in message for message in
caplog.messages)
+
@pytest.mark.asyncio
@mock.patch("airflow.providers.dbt.cloud.hooks.dbt.DbtCloudHook.get_job_status")
async def test_dbt_run_job_trigger(self, mock_get_job_status, end_time):
