jroachgolf84 commented on code in PR #69382:
URL: https://github.com/apache/airflow/pull/69382#discussion_r3610487922
##########
providers/tableau/src/airflow/providers/tableau/operators/tableau.py:
##########
@@ -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
Review Comment:
Quick question from my side. So, the behavior would be something like this:
- Task begins to execute
- Immediately, a `409093 Resource Conflict` is raised
- Rather than failing, the Task is skipped
Is that correct?
##########
providers/tableau/tests/unit/tableau/operators/test_tableau.py:
##########
@@ -530,3 +535,122 @@ def
test_blocking_refresh_forwards_wait_for_state_options(self, mock_tableau_hoo
exponential_backoff=True,
max_check_interval=120,
)
+
+ @patch("airflow.providers.tableau.operators.tableau.TableauHook")
+ def test_skip_on_conflict_skips_task(self, mock_tableau_hook):
+ """
+ Test that a 409093 Resource Conflict 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_conflict_not_skipped_by_default(self, mock_tableau_hook):
+ """
+ Test that a 409093 Resource Conflict 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_skip_on_conflict_does_not_swallow_other_server_errors(self,
mock_tableau_hook):
+ """
+ Test that skip_on_conflict 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_skip_on_conflict_skips_task_run(self, mock_tableau_hook):
+ """
+ Test that a 409093 Resource Conflict from tasks.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 from tasks.run still fails the
task when
Review Comment:
Does `tasks.run` refer to an Airflow Task? Or something else?
##########
providers/tableau/src/airflow/providers/tableau/operators/tableau.py:
##########
@@ -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
Review Comment:
Just thought I'd throw this out there. More generally, I'd also recommend to
use something like a pool/Worker Queue to handle something like this (to ensure
that only a single refresh for a resource can run). That's outside the scope of
this PR, but thought I'd mention it.
##########
providers/tableau/tests/unit/tableau/operators/test_tableau.py:
##########
@@ -530,3 +535,122 @@ def
test_blocking_refresh_forwards_wait_for_state_options(self, mock_tableau_hoo
exponential_backoff=True,
max_check_interval=120,
)
+
+ @patch("airflow.providers.tableau.operators.tableau.TableauHook")
+ def test_skip_on_conflict_skips_task(self, mock_tableau_hook):
+ """
+ Test that a 409093 Resource Conflict 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_conflict_not_skipped_by_default(self, mock_tableau_hook):
+ """
+ Test that a 409093 Resource Conflict 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_skip_on_conflict_does_not_swallow_other_server_errors(self,
mock_tableau_hook):
+ """
+ Test that skip_on_conflict 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_skip_on_conflict_skips_task_run(self, mock_tableau_hook):
Review Comment:
What is the difference between this test and
`test_skip_on_conflict_skips_task`?
##########
providers/tableau/tests/unit/tableau/operators/test_tableau.py:
##########
@@ -530,3 +535,122 @@ def
test_blocking_refresh_forwards_wait_for_state_options(self, mock_tableau_hoo
exponential_backoff=True,
max_check_interval=120,
)
+
+ @patch("airflow.providers.tableau.operators.tableau.TableauHook")
+ def test_skip_on_conflict_skips_task(self, mock_tableau_hook):
+ """
+ Test that a 409093 Resource Conflict 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_conflict_not_skipped_by_default(self, mock_tableau_hook):
+ """
+ Test that a 409093 Resource Conflict 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_skip_on_conflict_does_not_swallow_other_server_errors(self,
mock_tableau_hook):
+ """
+ Test that skip_on_conflict 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_skip_on_conflict_skips_task_run(self, mock_tableau_hook):
Review Comment:
It it's just the `run` method, can you denote that with a comment?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
