This is an automated email from the ASF dual-hosted git repository.
potiuk 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 688c25b7a26 Skip Tableau refresh/task run on 409093 resource conflict
(#69382)
688c25b7a26 is described below
commit 688c25b7a262e948500c91e8d6b0215e3576f999
Author: Jaya Haryono Manik <[email protected]>
AuthorDate: Thu Aug 13 20:16:05 2026 +0700
Skip Tableau refresh/task run on 409093 resource conflict (#69382)
Add `skip_on_conflict` support and improve conflict messages.
---
providers/tableau/docs/operators.rst | 1 +
.../airflow/providers/tableau/operators/tableau.py | 58 ++++++--
.../tests/unit/tableau/operators/test_tableau.py | 148 ++++++++++++++++++++-
3 files changed, 192 insertions(+), 15 deletions(-)
diff --git a/providers/tableau/docs/operators.rst
b/providers/tableau/docs/operators.rst
index 9bcc830b761..ab6141ce3dd 100644
--- a/providers/tableau/docs/operators.rst
+++ b/providers/tableau/docs/operators.rst
@@ -39,6 +39,7 @@ Using the Operator
| **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)
+| **skip_on_conflict**: When ``True``, treat a Tableau ``409093 Resource
Conflict`` error while triggering a refresh or running an extract refresh task
(a refresh/run for the same resource is already queued or running) as a skipped
task instead of a failure. Applies when ``method="refresh"`` and when
``method="run"`` on ``tasks``. **bool** - Default: **False**
| **tableau_conn_id**: The credentials to authenticate to the Tableau Server.
**str** - Default: **tableau_default**
|
|
diff --git
a/providers/tableau/src/airflow/providers/tableau/operators/tableau.py
b/providers/tableau/src/airflow/providers/tableau/operators/tableau.py
index f7158f95e42..bcf09798f7c 100644
--- a/providers/tableau/src/airflow/providers/tableau/operators/tableau.py
+++ b/providers/tableau/src/airflow/providers/tableau/operators/tableau.py
@@ -19,11 +19,12 @@ from __future__ import annotations
from collections.abc import Sequence
from typing import TYPE_CHECKING
-from tableauserverclient import JobItem
+from tableauserverclient import JobItem, ServerResponseError
from airflow.providers.common.compat.sdk import (
AirflowException,
AirflowOptionalProviderFeatureException,
+ AirflowSkipException,
BaseOperator,
)
from airflow.providers.tableau.hooks.tableau import (
@@ -35,6 +36,12 @@ from airflow.providers.tableau.hooks.tableau import (
if TYPE_CHECKING:
from airflow.providers.common.compat.sdk import Context
+# Tableau REST API error code returned when an extract refresh -- triggered
directly via
+# datasources/workbooks "refresh" or via running a scheduled extract refresh
task ("tasks.run")
+# -- is requested for a resource that already has one queued or running.
+# See:
https://help.tableau.com/current/api/rest_api/en-us/REST/rest_api_concepts_errors.htm
+RESOURCE_CONFLICT_ERROR_CODE = "409093"
+
RESOURCES_METHODS = {
"datasources": ["delete", "refresh"],
"groups": ["delete"],
@@ -75,6 +82,11 @@ class TableauOperator(BaseOperator):
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 skip_on_conflict: When ``True``, treat a Tableau ``409093 Resource
Conflict`` error
+ while triggering a refresh, or while running an extract refresh task,
as a skipped task
+ instead of a failure. Raised when a refresh/run for the same resource
is already queued
+ or running. Applies to ``method="refresh"`` and to ``method="run"`` on
``tasks``.
+ Defaults to ``False``, so the conflict fails the task as before.
:param tableau_conn_id: The :ref:`Tableau Connection id
<howto/connection:tableau>`
containing the credentials to authenticate to the Tableau Server.
"""
@@ -98,6 +110,7 @@ class TableauOperator(BaseOperator):
exponential_backoff: bool = False,
max_check_interval: float | None = None,
incremental_refresh: bool = False,
+ skip_on_conflict: bool = False,
tableau_conn_id: str = "tableau_default",
**kwargs,
) -> None:
@@ -113,6 +126,7 @@ class TableauOperator(BaseOperator):
self.site_id = site_id
self.blocking_refresh = blocking_refresh
self.incremental_refresh = incremental_refresh
+ self.skip_on_conflict = skip_on_conflict
self.tableau_conn_id = tableau_conn_id
def execute(self, context: Context) -> str:
@@ -147,24 +161,40 @@ class TableauOperator(BaseOperator):
if self.resource == "tasks" and self.method == "run":
task_item = resource.get_by_id(resource_id)
- response_bytes = method(task_item)
+ try:
+ response_bytes = method(task_item)
+ except ServerResponseError as e:
+ if self.skip_on_conflict and e.code ==
RESOURCE_CONFLICT_ERROR_CODE:
+ raise AirflowSkipException(
+ f"Tableau task {resource_id} run is already queued
or running "
+ f"({e.code}: {e.summary}); skipping this task."
+ ) from e
+ raise
job_items = JobItem.from_response(response_bytes,
tableau_hook.server.namespace)
if not job_items:
raise ValueError("Tableau tasks.run returned no JobItem in
response")
job_id = job_items[0].id
elif self.method == "refresh":
- if self.incremental_refresh:
- try:
- response = method(resource_id, incremental=True)
- except TypeError as e:
- if "incremental" in str(e):
- raise AirflowOptionalProviderFeatureException(
- "Incremental refresh requires
tableauserverclient>=0.35. "
- "Please upgrade: pip install
'tableauserverclient>=0.35'"
- ) from e
- raise
- else:
- response = method(resource_id)
+ try:
+ if self.incremental_refresh:
+ try:
+ response = method(resource_id, incremental=True)
+ except TypeError as e:
+ if "incremental" in str(e):
+ raise AirflowOptionalProviderFeatureException(
+ "Incremental refresh requires
tableauserverclient>=0.35. "
+ "Please upgrade: pip install
'tableauserverclient>=0.35'"
+ ) from e
+ raise
+ else:
+ response = method(resource_id)
+ except ServerResponseError as e:
+ if self.skip_on_conflict and e.code ==
RESOURCE_CONFLICT_ERROR_CODE:
+ raise AirflowSkipException(
+ f"Tableau {self.resource} refresh is already
queued or running "
+ f"({e.code}: {e.summary}); skipping this task."
+ ) from e
+ raise
job_id = response.id
else:
response = method(resource_id)
diff --git a/providers/tableau/tests/unit/tableau/operators/test_tableau.py
b/providers/tableau/tests/unit/tableau/operators/test_tableau.py
index 759a0344c01..6b27de71d1e 100644
--- a/providers/tableau/tests/unit/tableau/operators/test_tableau.py
+++ b/providers/tableau/tests/unit/tableau/operators/test_tableau.py
@@ -19,8 +19,13 @@ from __future__ import annotations
from unittest.mock import Mock, patch
import pytest
+from tableauserverclient import ServerResponseError
-from airflow.providers.common.compat.sdk import AirflowException,
AirflowOptionalProviderFeatureException
+from airflow.providers.common.compat.sdk import (
+ AirflowException,
+ AirflowOptionalProviderFeatureException,
+ AirflowSkipException,
+)
from airflow.providers.tableau.hooks.tableau import TableauJobFinishCode
from airflow.providers.tableau.operators.tableau import TableauOperator
@@ -530,3 +535,144 @@ class TestTableauOperator:
exponential_backoff=True,
max_check_interval=120,
)
+
+ @patch("airflow.providers.tableau.operators.tableau.TableauHook")
+ def test_resource_refresh_skip_on_conflict_raises_skip_exception(self,
mock_tableau_hook):
+ """
+ Test that a 409093 Resource Conflict on a Tableau resource refresh is
turned into an
+ AirflowSkipException when skip_on_conflict=True.
+ """
+ mock_tableau_hook.get_all = Mock(return_value=self.mock_datasources)
+ mock_tableau_hook.return_value.__enter__ =
Mock(return_value=mock_tableau_hook)
+ mock_tableau_hook.server.datasources.refresh.side_effect =
ServerResponseError(
+ "409093", "Resource Conflict", "Job is already queued. Not queuing
a duplicate."
+ )
+
+ operator = TableauOperator(
+ find="ds_2",
+ resource="datasources",
+ skip_on_conflict=True,
+ **self.kwargs,
+ )
+
+ with pytest.raises(AirflowSkipException):
+ operator.execute(context={})
+
+ @patch("airflow.providers.tableau.operators.tableau.TableauHook")
+ def
test_incremental_resource_refresh_skip_on_conflict_raises_skip_exception(self,
mock_tableau_hook):
+ """
+ Test that a 409093 Resource Conflict on a Tableau incremental resource
refresh is turned into an
+ AirflowSkipException when skip_on_conflict=True.
+ """
+ mock_tableau_hook.get_all = Mock(return_value=self.mock_datasources)
+ mock_tableau_hook.return_value.__enter__ =
Mock(return_value=mock_tableau_hook)
+ mock_tableau_hook.server.datasources.refresh.side_effect =
ServerResponseError(
+ "409093", "Resource Conflict", "Job is already queued. Not queuing
a duplicate."
+ )
+ operator = TableauOperator(
+ find="ds_2",
+ resource="datasources",
+ incremental_refresh=True,
+ skip_on_conflict=True,
+ **self.kwargs,
+ )
+ with pytest.raises(AirflowSkipException):
+ operator.execute(context={})
+
+ @patch("airflow.providers.tableau.operators.tableau.TableauHook")
+ def test_resource_refresh_conflict_not_skipped_by_default(self,
mock_tableau_hook):
+ """
+ Test that a 409093 Resource Conflict on a Tableau resource refresh
still fails the
+ task when skip_on_conflict is left at its default (False), preserving
pre-existing
+ behavior.
+ """
+ mock_tableau_hook.get_all = Mock(return_value=self.mock_datasources)
+ mock_tableau_hook.return_value.__enter__ =
Mock(return_value=mock_tableau_hook)
+ mock_tableau_hook.server.datasources.refresh.side_effect =
ServerResponseError(
+ "409093", "Resource Conflict", "Job is already queued. Not queuing
a duplicate."
+ )
+
+ operator = TableauOperator(
+ find="ds_2",
+ resource="datasources",
+ **self.kwargs,
+ )
+
+ with pytest.raises(ServerResponseError):
+ operator.execute(context={})
+
+ @patch("airflow.providers.tableau.operators.tableau.TableauHook")
+ def
test_resource_refresh_skip_on_conflict_does_not_swallow_other_server_errors(self,
mock_tableau_hook):
+ """
+ Test that skip_on_conflict on a Tableau resource refresh only catches
the 409093
+ conflict code; other ServerResponseErrors still fail the task.
+ """
+ mock_tableau_hook.get_all = Mock(return_value=self.mock_datasources)
+ mock_tableau_hook.return_value.__enter__ =
Mock(return_value=mock_tableau_hook)
+ mock_tableau_hook.server.datasources.refresh.side_effect =
ServerResponseError(
+ "500000", "Internal Server Error", "Something else went wrong."
+ )
+
+ operator = TableauOperator(
+ find="ds_2",
+ resource="datasources",
+ skip_on_conflict=True,
+ **self.kwargs,
+ )
+
+ with pytest.raises(ServerResponseError):
+ operator.execute(context={})
+
+ @patch("airflow.providers.tableau.operators.tableau.TableauHook")
+ def test_task_run_skip_on_conflict_raises_skip_exception(self,
mock_tableau_hook):
+ """
+ Test that a 409093 Resource Conflict on a Tableau task run is turned
into an
+ AirflowSkipException when skip_on_conflict=True.
+ """
+ mock_tableau_hook.return_value.__enter__ =
Mock(return_value=mock_tableau_hook)
+ mock_tableau_hook.server.tasks.get_by_id = Mock(return_value=Mock())
+ mock_tableau_hook.server.tasks.run.side_effect = ServerResponseError(
+ "409093", "Resource Conflict", "Job is already queued. Not queuing
a duplicate."
+ )
+
+ kwargs = {**self.kwargs, "method": "run", "match_with": "id"}
+ operator = TableauOperator(find="task-abc", resource="tasks",
skip_on_conflict=True, **kwargs)
+
+ with pytest.raises(AirflowSkipException):
+ operator.execute(context={})
+
+ @patch("airflow.providers.tableau.operators.tableau.TableauHook")
+ def test_task_run_conflict_not_skipped_by_default(self, mock_tableau_hook):
+ """
+ Test that a 409093 Resource Conflict on a Tableau task run still fails
the task when
+ skip_on_conflict is left at its default (False), preserving
pre-existing behavior.
+ """
+ mock_tableau_hook.return_value.__enter__ =
Mock(return_value=mock_tableau_hook)
+ mock_tableau_hook.server.tasks.get_by_id = Mock(return_value=Mock())
+ mock_tableau_hook.server.tasks.run.side_effect = ServerResponseError(
+ "409093", "Resource Conflict", "Job is already queued. Not queuing
a duplicate."
+ )
+
+ kwargs = {**self.kwargs, "method": "run", "match_with": "id"}
+ operator = TableauOperator(find="task-abc", resource="tasks", **kwargs)
+
+ with pytest.raises(ServerResponseError):
+ operator.execute(context={})
+
+ @patch("airflow.providers.tableau.operators.tableau.TableauHook")
+ def
test_task_run_skip_on_conflict_does_not_swallow_other_server_errors(self,
mock_tableau_hook):
+ """
+ Test that skip_on_conflict on a Tableau task run only catches the
409093 conflict
+ code; other ServerResponseErrors still fail the task.
+ """
+ mock_tableau_hook.return_value.__enter__ =
Mock(return_value=mock_tableau_hook)
+ mock_tableau_hook.server.tasks.get_by_id = Mock(return_value=Mock())
+ mock_tableau_hook.server.tasks.run.side_effect = ServerResponseError(
+ "500000", "Internal Server Error", "Something else went wrong."
+ )
+
+ kwargs = {**self.kwargs, "method": "run", "match_with": "id"}
+ operator = TableauOperator(find="task-abc", resource="tasks",
skip_on_conflict=True, **kwargs)
+
+ with pytest.raises(ServerResponseError):
+ operator.execute(context={})
