This is an automated email from the ASF dual-hosted git repository.
kaxil 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 5e65f218508 Fail the Teradata compute cluster trigger on an unknown
operation (#72696)
5e65f218508 is described below
commit 5e65f218508e7249fb562d10186c4ceb0a04218b
Author: rjgoyln <[email protected]>
AuthorDate: Wed Sep 9 19:09:53 2026 +0800
Fail the Teradata compute cluster trigger on an unknown operation (#72696)
* Fail the Teradata compute cluster trigger on an unknown operation
The polling loop only exits once the compute profile reaches the status the
operation waits for. An operation type the trigger does not recognise never
leaves it, so the trigger polls on instead of reporting the bad input.
No shipped operator passes such a value, so this is hardening rather than a
fault reachable from a Dag. The "Invalid operation" event meant to cover the
case sat after the loop, unreachable, as did the two timeout events beside
it -
the triggerer's own deferral timeout has always superseded those.
* Name the rejected Teradata operation type and correct its docstring
The operator surfaces the trigger's error message verbatim as the task's
AirflowException, so a bare "Invalid operation" gave no way to tell which
value
was rejected. The docstring is now the only place the accepted values are
written down, and it named a parameter that does not exist while omitting
the
two CREATE operations.
---
generated/known_airflow_exceptions.txt | 1 -
.../teradata/triggers/teradata_compute_cluster.py | 77 ++++++----------------
.../triggers/test_teradata_compute_cluster.py | 43 ++++++++++++
3 files changed, 62 insertions(+), 59 deletions(-)
diff --git a/generated/known_airflow_exceptions.txt
b/generated/known_airflow_exceptions.txt
index cd9b8e3c99c..2c619d8eaff 100644
--- a/generated/known_airflow_exceptions.txt
+++ b/generated/known_airflow_exceptions.txt
@@ -400,7 +400,6 @@
providers/telegram/src/airflow/providers/telegram/operators/telegram.py::2
providers/teradata/src/airflow/providers/teradata/hooks/bteq.py::10
providers/teradata/src/airflow/providers/teradata/hooks/ttu.py::1
providers/teradata/src/airflow/providers/teradata/operators/teradata_compute_cluster.py::9
-providers/teradata/src/airflow/providers/teradata/triggers/teradata_compute_cluster.py::1
providers/teradata/src/airflow/providers/teradata/utils/bteq_util.py::2
providers/trino/src/airflow/providers/trino/hooks/trino.py::1
providers/vespa/src/airflow/providers/vespa/operators/vespa_ingest.py::1
diff --git
a/providers/teradata/src/airflow/providers/teradata/triggers/teradata_compute_cluster.py
b/providers/teradata/src/airflow/providers/teradata/triggers/teradata_compute_cluster.py
index f9c90313d83..9b31997bbc8 100644
---
a/providers/teradata/src/airflow/providers/teradata/triggers/teradata_compute_cluster.py
+++
b/providers/teradata/src/airflow/providers/teradata/triggers/teradata_compute_cluster.py
@@ -20,7 +20,6 @@ import asyncio
from collections.abc import AsyncIterator
from typing import Any
-from airflow.providers.common.compat.sdk import AirflowException
from airflow.providers.common.sql.hooks.handlers import fetch_one_handler
from airflow.providers.teradata.hooks.teradata import TeradataHook
from airflow.providers.teradata.utils.constants import Constants
@@ -29,13 +28,13 @@ from airflow.triggers.base import BaseTrigger, TriggerEvent
class TeradataComputeClusterSyncTrigger(BaseTrigger):
"""
- Fetch the status of the suspend or resume operation for the specified
compute cluster.
+ Fetch the status of the requested operation for the specified compute
cluster.
:param teradata_conn_id: The :ref:`Teradata connection id
<howto/connection:teradata>`
reference to a specific Teradata database.
:param compute_profile_name: Name of the Compute Profile to manage.
:param compute_group_name: Name of compute group to which compute profile
belongs.
- :param opr_type: Compute cluster operation - SUSPEND/RESUME
+ :param operation_type: Compute cluster operation -
CREATE/CREATE_SUSPEND/RESUME/SUSPEND
:param poll_interval: polling period in minutes to check for the status
"""
@@ -69,70 +68,32 @@ class TeradataComputeClusterSyncTrigger(BaseTrigger):
async def run(self) -> AsyncIterator[TriggerEvent]:
"""Wait for Compute Cluster operation to complete."""
+ if self.operation_type in (Constants.CC_SUSPEND_OPR,
Constants.CC_CREATE_SUSPEND_OPR):
+ expected_status = Constants.CC_SUSPEND_DB_STATUS
+ elif self.operation_type in (Constants.CC_RESUME_OPR,
Constants.CC_CREATE_OPR):
+ expected_status = Constants.CC_RESUME_DB_STATUS
+ else:
+ yield TriggerEvent({"status": "error", "message": f"Invalid
operation: {self.operation_type}"})
+ return
try:
while True:
status = await self.get_status()
if status is None or len(status) == 0:
- raise AirflowException(Constants.CC_GRP_PRP_NON_EXISTS_MSG
% "manage")
- if (
- self.operation_type == Constants.CC_SUSPEND_OPR
- or self.operation_type == Constants.CC_CREATE_SUSPEND_OPR
- ):
- if status == Constants.CC_SUSPEND_DB_STATUS:
- break
- elif (
- self.operation_type == Constants.CC_RESUME_OPR
- or self.operation_type == Constants.CC_CREATE_OPR
- ):
- if status == Constants.CC_RESUME_DB_STATUS:
- break
+ raise ValueError(Constants.CC_GRP_PRP_NON_EXISTS_MSG %
"manage")
+ if status == expected_status:
+ break
if self.poll_interval is not None:
self.poll_interval = float(self.poll_interval)
else:
self.poll_interval = float(Constants.CC_POLL_INTERVAL)
await asyncio.sleep(self.poll_interval)
- if (
- self.operation_type == Constants.CC_SUSPEND_OPR
- or self.operation_type == Constants.CC_CREATE_SUSPEND_OPR
- ):
- if status == Constants.CC_SUSPEND_DB_STATUS:
- yield TriggerEvent(
- {
- "status": "success",
- "message": Constants.CC_OPR_SUCCESS_STATUS_MSG
- % (self.compute_profile_name, self.operation_type),
- }
- )
- else:
- yield TriggerEvent(
- {
- "status": "error",
- "message": Constants.CC_OPR_TIMEOUT_ERROR
- % (self.operation_type, self.compute_profile_name),
- }
- )
- elif (
- self.operation_type == Constants.CC_RESUME_OPR
- or self.operation_type == Constants.CC_CREATE_OPR
- ):
- if status == Constants.CC_RESUME_DB_STATUS:
- yield TriggerEvent(
- {
- "status": "success",
- "message": Constants.CC_OPR_SUCCESS_STATUS_MSG
- % (self.compute_profile_name, self.operation_type),
- }
- )
- else:
- yield TriggerEvent(
- {
- "status": "error",
- "message": Constants.CC_OPR_TIMEOUT_ERROR
- % (self.operation_type, self.compute_profile_name),
- }
- )
- else:
- yield TriggerEvent({"status": "error", "message": "Invalid
operation"})
+ yield TriggerEvent(
+ {
+ "status": "success",
+ "message": Constants.CC_OPR_SUCCESS_STATUS_MSG
+ % (self.compute_profile_name, self.operation_type),
+ }
+ )
except Exception as e:
yield TriggerEvent({"status": "error", "message": str(e)})
except asyncio.CancelledError:
diff --git
a/providers/teradata/tests/unit/teradata/triggers/test_teradata_compute_cluster.py
b/providers/teradata/tests/unit/teradata/triggers/test_teradata_compute_cluster.py
index a9a1fedce6c..ee454e5c4eb 100644
---
a/providers/teradata/tests/unit/teradata/triggers/test_teradata_compute_cluster.py
+++
b/providers/teradata/tests/unit/teradata/triggers/test_teradata_compute_cluster.py
@@ -125,6 +125,49 @@ async def test_run_resume_failure():
mock_get_status.assert_called_once()
[email protected]
+async def test_run_invalid_operation():
+ trigger = TeradataComputeClusterSyncTrigger(
+ teradata_conn_id="test_conn_id",
+ compute_profile_name="test_profile",
+ operation_type="INVALID",
+ poll_interval=1,
+ )
+ with patch.object(trigger, "get_status", autospec=True) as mock_get_status:
+ events = [event async for event in trigger.run()]
+ assert events == [TriggerEvent({"status": "error", "message": "Invalid
operation: INVALID"})]
+ mock_get_status.assert_not_called()
+
+
[email protected]
[email protected](
+ ("operation_type", "db_status"),
+ [
+ (Constants.CC_CREATE_OPR, Constants.CC_RESUME_DB_STATUS),
+ (Constants.CC_CREATE_SUSPEND_OPR, Constants.CC_SUSPEND_DB_STATUS),
+ ],
+)
+async def test_run_create_success(operation_type, db_status):
+ trigger = TeradataComputeClusterSyncTrigger(
+ teradata_conn_id="test_conn_id",
+ compute_profile_name="test_profile",
+ operation_type=operation_type,
+ poll_interval=1,
+ )
+ with patch.object(trigger, "get_status", autospec=True) as mock_get_status:
+ mock_get_status.return_value = db_status
+ events = [event async for event in trigger.run()]
+ assert events == [
+ TriggerEvent(
+ {
+ "status": "success",
+ "message": Constants.CC_OPR_SUCCESS_STATUS_MSG %
("test_profile", operation_type),
+ }
+ )
+ ]
+ mock_get_status.assert_called_once()
+
+
@pytest.fixture
def mock_teradata_hook_run():
with patch.object(TeradataHook, "run") as mock_run:
