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 fd5686c8320 Add timeout and exponential backoff to Tableau refresh
wait (#69479)
fd5686c8320 is described below
commit fd5686c83200918de9183eb6b38873dc0dcbf018
Author: Jaya Haryono Manik <[email protected]>
AuthorDate: Tue Jul 7 01:33:00 2026 +0700
Add timeout and exponential backoff to Tableau refresh wait (#69479)
---
providers/tableau/docs/operators.rst | 3 +
.../src/airflow/providers/tableau/hooks/tableau.py | 36 ++++++-
.../airflow/providers/tableau/operators/tableau.py | 16 +++
.../tests/unit/tableau/hooks/test_tableau.py | 115 ++++++++++++++++++++-
.../tests/unit/tableau/operators/test_tableau.py | 55 ++++++++--
5 files changed, 213 insertions(+), 12 deletions(-)
diff --git a/providers/tableau/docs/operators.rst
b/providers/tableau/docs/operators.rst
index f4aac22e91a..9bcc830b761 100644
--- a/providers/tableau/docs/operators.rst
+++ b/providers/tableau/docs/operators.rst
@@ -36,6 +36,9 @@ Using the Operator
| **site_id**: The id of the site where the workbook belongs to. **str** -
Default: **None**
| **blocking_refresh**: By default the extract refresh will be blocking means
it will wait until it has finished. **bool** - Default: **True**
| **check_interval**: time in seconds that the job should wait in between each
instance state checks until operation is completed. **float** - Default: **20**
+| **timeout**: maximum total time in seconds to wait for a blocking refresh
before giving up with a ``TimeoutError``. **float** - Default: **None** (wait
indefinitely)
+| **exponential_backoff**: grow the wait between status checks by 50% each
time, starting from ``check_interval``, instead of staying fixed. **bool** -
Default: **False**
+| **max_check_interval**: maximum interval in seconds between two consecutive
status checks when ``exponential_backoff`` is enabled. **float** - Default:
**None** (uncapped)
| **tableau_conn_id**: The credentials to authenticate to the Tableau Server.
**str** - Default: **tableau_default**
|
|
diff --git a/providers/tableau/src/airflow/providers/tableau/hooks/tableau.py
b/providers/tableau/src/airflow/providers/tableau/hooks/tableau.py
index 9aa994da9ec..b603d338fad 100644
--- a/providers/tableau/src/airflow/providers/tableau/hooks/tableau.py
+++ b/providers/tableau/src/airflow/providers/tableau/hooks/tableau.py
@@ -190,7 +190,15 @@ class TableauHook(BaseHook):
"""
return
TableauJobFinishCode(int(self.server.jobs.get_by_id(job_id).finish_code))
- def wait_for_state(self, job_id: str, target_state: TableauJobFinishCode,
check_interval: float) -> bool:
+ def wait_for_state(
+ self,
+ job_id: str,
+ target_state: TableauJobFinishCode,
+ check_interval: float,
+ timeout: float | None = None,
+ exponential_backoff: bool = False,
+ max_check_interval: float | None = None,
+ ) -> bool:
"""
Wait until the current state of a defined Tableau Job is target_state
or different from PENDING.
@@ -198,12 +206,30 @@ class TableauHook(BaseHook):
:param target_state: Enum that describe the Tableau job's target state
:param check_interval: time in seconds that the job should wait in
between each instance state checks until operation is completed
+ :param timeout: maximum total time in seconds to keep polling the job
before giving up.
+ ``None`` (the default) waits indefinitely until the job leaves the
PENDING state.
+ This is a soft bound: an in-flight wait between checks is not
interrupted, so a large
+ ``check_interval`` may overshoot ``timeout`` by up to one interval.
+ :param exponential_backoff: when ``True`` the wait between checks
grows by 50% each time,
+ starting from ``check_interval``, instead of staying fixed.
+ :param max_check_interval: maximum interval in seconds between two
consecutive checks
+ when ``exponential_backoff`` is enabled. ``None`` leaves the
growth uncapped.
:return: return True if the job is equal to the target_status, False
otherwise.
+ :raises TimeoutError: if ``timeout`` elapses while the job is still
PENDING.
"""
- finish_code = self.get_job_status(job_id=job_id)
- while finish_code == TableauJobFinishCode.PENDING and finish_code !=
target_state:
- self.log.info("job state: %s", finish_code)
- time.sleep(check_interval)
+ start = time.monotonic()
+ current_interval = check_interval
+ while True:
finish_code = self.get_job_status(job_id=job_id)
+ if finish_code != TableauJobFinishCode.PENDING or finish_code ==
target_state:
+ break
+ if timeout is not None and time.monotonic() - start >= timeout:
+ raise TimeoutError(f"Tableau job {job_id} is still PENDING
after {timeout} seconds.")
+ self.log.info("job state: %s; checking again in %s seconds",
finish_code, current_interval)
+ time.sleep(current_interval)
+ if exponential_backoff:
+ current_interval *= 1.5
+ if max_check_interval is not None:
+ current_interval = min(current_interval,
max_check_interval)
return finish_code == target_state
diff --git
a/providers/tableau/src/airflow/providers/tableau/operators/tableau.py
b/providers/tableau/src/airflow/providers/tableau/operators/tableau.py
index 9cb58657645..f7158f95e42 100644
--- a/providers/tableau/src/airflow/providers/tableau/operators/tableau.py
+++ b/providers/tableau/src/airflow/providers/tableau/operators/tableau.py
@@ -66,6 +66,13 @@ class TableauOperator(BaseOperator):
:param blocking_refresh: By default will be blocking means it will wait
until it has finished.
:param check_interval: time in seconds that the job should wait in
between each instance state checks until operation is completed
+ :param timeout: maximum total time in seconds to wait for a blocking
refresh to finish before
+ giving up with a ``TimeoutError``. ``None`` (the default) waits
indefinitely until the job
+ leaves the PENDING state.
+ :param exponential_backoff: when ``True`` the wait between status checks
grows by 50% each
+ time, starting from ``check_interval``, instead of staying fixed.
+ :param max_check_interval: maximum interval in seconds between two
consecutive status checks
+ when ``exponential_backoff`` is enabled. ``None`` leaves the growth
uncapped.
:param incremental_refresh: Whether to perform an incremental refresh
instead of a full refresh.
Only applies to datasource and workbook refresh operations. Defaults
to False (full refresh).
:param tableau_conn_id: The :ref:`Tableau Connection id
<howto/connection:tableau>`
@@ -87,6 +94,9 @@ class TableauOperator(BaseOperator):
site_id: str | None = None,
blocking_refresh: bool = True,
check_interval: float = 20,
+ timeout: float | None = None,
+ exponential_backoff: bool = False,
+ max_check_interval: float | None = None,
incremental_refresh: bool = False,
tableau_conn_id: str = "tableau_default",
**kwargs,
@@ -97,6 +107,9 @@ class TableauOperator(BaseOperator):
self.find = find
self.match_with = match_with
self.check_interval = check_interval
+ self.timeout = timeout
+ self.exponential_backoff = exponential_backoff
+ self.max_check_interval = max_check_interval
self.site_id = site_id
self.blocking_refresh = blocking_refresh
self.incremental_refresh = incremental_refresh
@@ -163,6 +176,9 @@ class TableauOperator(BaseOperator):
job_id=job_id,
check_interval=self.check_interval,
target_state=TableauJobFinishCode.SUCCESS,
+ timeout=self.timeout,
+ exponential_backoff=self.exponential_backoff,
+ max_check_interval=self.max_check_interval,
):
raise TableauJobFailedException(f"The Tableau Refresh
{self.resource} Job failed!")
diff --git a/providers/tableau/tests/unit/tableau/hooks/test_tableau.py
b/providers/tableau/tests/unit/tableau/hooks/test_tableau.py
index 977dbf07558..3b9b084ee8d 100644
--- a/providers/tableau/tests/unit/tableau/hooks/test_tableau.py
+++ b/providers/tableau/tests/unit/tableau/hooks/test_tableau.py
@@ -18,7 +18,7 @@ from __future__ import annotations
import json
import tempfile
-from unittest.mock import MagicMock, patch
+from unittest.mock import MagicMock, call, patch
import pytest
@@ -460,3 +460,116 @@ class TestTableauHook:
assert tableau_hook.wait_for_state(
job_id="j1", target_state=TableauJobFinishCode.PENDING,
check_interval=1
)
+
+ @patch("tableauserverclient.Server")
+ def test_wait_for_state_timeout(self, mock_tableau_server):
+ """A job stuck in PENDING raises TimeoutError once the timeout
elapses."""
+ clock = {"t": 0.0}
+ with TableauHook(tableau_conn_id="tableau_test_password") as
tableau_hook:
+ tableau_hook.get_job_status = MagicMock(
+ name="get_job_status",
+ return_value=TableauJobFinishCode.PENDING,
+ )
+ with (
+ patch("time.monotonic", side_effect=lambda: clock["t"]),
+ patch("time.sleep", side_effect=lambda seconds:
clock.__setitem__("t", clock["t"] + seconds)),
+ ):
+ with pytest.raises(TimeoutError, match="still PENDING after 30
seconds"):
+ tableau_hook.wait_for_state(
+ job_id="j1",
+ target_state=TableauJobFinishCode.SUCCESS,
+ check_interval=10,
+ timeout=30,
+ )
+
+ @pytest.mark.parametrize(
+ ("job_status", "expected"),
+ [
+ pytest.param(
+ MagicMock(
+ name="get_job_status",
+ side_effect=[TableauJobFinishCode.PENDING,
TableauJobFinishCode.SUCCESS],
+ ),
+ True,
+ id="finished-at-deadline",
+ ),
+ pytest.param(
+ MagicMock(name="get_job_status",
return_value=TableauJobFinishCode.PENDING),
+ "raises",
+ id="still-pending",
+ ),
+ ],
+ )
+ @patch("tableauserverclient.Server")
+ def test_wait_for_state_uses_last_poll_result_on_timeout(self,
mock_tableau_server, job_status, expected):
+ """A job that finishes on the poll taken at the deadline succeeds; one
still PENDING times out."""
+ clock = {"t": 0.0}
+ with TableauHook(tableau_conn_id="tableau_test_password") as
tableau_hook:
+ tableau_hook.get_job_status = job_status
+ with (
+ patch("time.monotonic", side_effect=lambda: clock["t"]),
+ patch("time.sleep", side_effect=lambda seconds:
clock.__setitem__("t", clock["t"] + seconds)),
+ ):
+ if expected == "raises":
+ with pytest.raises(TimeoutError):
+ tableau_hook.wait_for_state(
+ job_id="j1",
+ target_state=TableauJobFinishCode.SUCCESS,
+ check_interval=5,
+ timeout=5,
+ )
+ else:
+ assert tableau_hook.wait_for_state(
+ job_id="j1",
+ target_state=TableauJobFinishCode.SUCCESS,
+ check_interval=5,
+ timeout=5,
+ )
+
+ @patch("time.sleep", return_value=None)
+ @patch("tableauserverclient.Server")
+ def test_wait_for_state_timeout_success_before_elapsed(self,
mock_tableau_server, sleep_mock):
+ """timeout does not interfere when the job finishes before it
elapses."""
+ with TableauHook(tableau_conn_id="tableau_test_password") as
tableau_hook:
+ tableau_hook.get_job_status = MagicMock(
+ name="get_job_status",
+ side_effect=[TableauJobFinishCode.PENDING,
TableauJobFinishCode.SUCCESS],
+ )
+ assert tableau_hook.wait_for_state(
+ job_id="j1",
+ target_state=TableauJobFinishCode.SUCCESS,
+ check_interval=1,
+ timeout=300,
+ )
+
+ @pytest.mark.parametrize(
+ ("max_check_interval", "expected_intervals"),
+ [
+ pytest.param(None, [10, 15, 22.5], id="uncapped"),
+ pytest.param(12, [10, 12, 12], id="capped"),
+ ],
+ )
+ @patch("time.sleep", return_value=None)
+ @patch("tableauserverclient.Server")
+ def test_wait_for_state_exponential_backoff(
+ self, mock_tableau_server, sleep_mock, max_check_interval,
expected_intervals
+ ):
+ """Exponential backoff grows the wait between checks by 50% each time,
honoring the cap."""
+ with TableauHook(tableau_conn_id="tableau_test_password") as
tableau_hook:
+ tableau_hook.get_job_status = MagicMock(
+ name="get_job_status",
+ side_effect=[
+ TableauJobFinishCode.PENDING,
+ TableauJobFinishCode.PENDING,
+ TableauJobFinishCode.PENDING,
+ TableauJobFinishCode.SUCCESS,
+ ],
+ )
+ assert tableau_hook.wait_for_state(
+ job_id="j1",
+ target_state=TableauJobFinishCode.SUCCESS,
+ check_interval=10,
+ exponential_backoff=True,
+ max_check_interval=max_check_interval,
+ )
+ assert sleep_mock.call_args_list == [call(interval) for interval
in expected_intervals]
diff --git a/providers/tableau/tests/unit/tableau/operators/test_tableau.py
b/providers/tableau/tests/unit/tableau/operators/test_tableau.py
index c2e9e5157fc..759a0344c01 100644
--- a/providers/tableau/tests/unit/tableau/operators/test_tableau.py
+++ b/providers/tableau/tests/unit/tableau/operators/test_tableau.py
@@ -87,7 +87,7 @@ class TestTableauOperator:
def mock_hook_exit(exc_type, exc_val, exc_tb):
mock_signed_in[0] = False
- def mock_wait_for_state(job_id, target_state, check_interval):
+ def mock_wait_for_state(job_id, target_state, check_interval,
**kwargs):
if not mock_signed_in[0]:
raise Exception("Not signed in")
@@ -109,7 +109,12 @@ class TestTableauOperator:
mock_tableau_hook.server.workbooks.refresh.assert_called_once_with(2)
assert mock_tableau_hook.server.workbooks.refresh.return_value.id ==
job_id
mock_tableau_hook.wait_for_state.assert_called_once_with(
- job_id=job_id, check_interval=20,
target_state=TableauJobFinishCode.SUCCESS
+ job_id=job_id,
+ check_interval=20,
+ target_state=TableauJobFinishCode.SUCCESS,
+ timeout=None,
+ exponential_backoff=False,
+ max_check_interval=None,
)
@patch("airflow.providers.tableau.operators.tableau.TableauHook")
@@ -152,7 +157,7 @@ class TestTableauOperator:
def mock_hook_exit(exc_type, exc_val, exc_tb):
mock_signed_in[0] = False
- def mock_wait_for_state(job_id, target_state, check_interval):
+ def mock_wait_for_state(job_id, target_state, check_interval,
**kwargs):
if not mock_signed_in[0]:
raise Exception("Not signed in")
@@ -170,7 +175,12 @@ class TestTableauOperator:
mock_tableau_hook.server.datasources.refresh.assert_called_once_with(2)
assert mock_tableau_hook.server.datasources.refresh.return_value.id ==
job_id
mock_tableau_hook.wait_for_state.assert_called_once_with(
- job_id=job_id, check_interval=20,
target_state=TableauJobFinishCode.SUCCESS
+ job_id=job_id,
+ check_interval=20,
+ target_state=TableauJobFinishCode.SUCCESS,
+ timeout=None,
+ exponential_backoff=False,
+ max_check_interval=None,
)
@patch("airflow.providers.tableau.operators.tableau.TableauHook")
@@ -372,7 +382,7 @@ class TestTableauOperator:
def mock_hook_exit(exc_type, exc_val, exc_tb):
mock_signed_in[0] = False
- def mock_wait_for_state(job_id, target_state, check_interval):
+ def mock_wait_for_state(job_id, target_state, check_interval,
**kwargs):
if not mock_signed_in[0]:
raise Exception("Not signed in")
return True
@@ -394,7 +404,12 @@ class TestTableauOperator:
mock_tableau_hook.server.datasources.refresh.assert_called_once_with(2,
incremental=True)
assert mock_tableau_hook.server.datasources.refresh.return_value.id ==
job_id
mock_tableau_hook.wait_for_state.assert_called_once_with(
- job_id=job_id, check_interval=20,
target_state=TableauJobFinishCode.SUCCESS
+ job_id=job_id,
+ check_interval=20,
+ target_state=TableauJobFinishCode.SUCCESS,
+ timeout=None,
+ exponential_backoff=False,
+ max_check_interval=None,
)
@patch("airflow.providers.tableau.operators.tableau.TableauHook")
@@ -487,3 +502,31 @@ class TestTableauOperator:
# Verify that refresh was called WITHOUT the incremental parameter
mock_tableau_hook.server.datasources.refresh.assert_called_once_with(2)
assert job_id ==
mock_tableau_hook.server.datasources.refresh.return_value.id
+
+ @patch("airflow.providers.tableau.operators.tableau.TableauHook")
+ def test_blocking_refresh_forwards_wait_for_state_options(self,
mock_tableau_hook):
+ """timeout/exponential_backoff/max_check_interval should be forwarded
to wait_for_state."""
+ mock_tableau_hook.return_value.__enter__ =
Mock(return_value=mock_tableau_hook)
+ mock_tableau_hook.wait_for_state = Mock(return_value=True)
+ mock_tableau_hook.get_all = Mock(return_value=self.mock_datasources)
+
+ operator = TableauOperator(
+ find="ds_2",
+ resource="datasources",
+ check_interval=5,
+ timeout=300,
+ exponential_backoff=True,
+ max_check_interval=120,
+ **self.kwargs,
+ )
+
+ job_id = operator.execute(context={})
+
+ mock_tableau_hook.wait_for_state.assert_called_once_with(
+ job_id=job_id,
+ check_interval=5,
+ target_state=TableauJobFinishCode.SUCCESS,
+ timeout=300,
+ exponential_backoff=True,
+ max_check_interval=120,
+ )
