Add 'ignore_if_missing' to DataprocDeleteClusterOperator

Author: shahar1

providers/google/src/airflow/providers/google/cloud/operators/dataproc.py

@@ -949,6 +949,8 @@ class DataprocDeleteClusterOperator(GoogleCloudBaseOperator):
         account from the list granting this role to the originating account (templated).
     :param deferrable: Run operator in the deferrable mode.
     :param polling_interval_seconds: Time (seconds) to wait between calls to check the cluster status.
+    :param ignore_if_missing: If True, the operator will not raise an exception if the cluster does not exist.
+        Defaults to False.
     """
 
     template_fields: Sequence[str] = (
@@ -974,11 +976,14 @@ def __init__(
         impersonation_chain: str | Sequence[str] | None = None,
         deferrable: bool = conf.getboolean("operators", "default_deferrable", fallback=False),
         polling_interval_seconds: int = 10,
+        ignore_if_missing: bool = False,
         **kwargs,
     ):
         super().__init__(**kwargs)
         if deferrable and polling_interval_seconds <= 0:
             raise ValueError("Invalid value for polling_interval_seconds. Expected value greater than 0")
+        if deferrable and ignore_if_missing:
+            raise ValueError("Cannot set both deferrable and ignore_if_missing to True")
         self.project_id = project_id
         self.region = region
         self.cluster_name = cluster_name
@@ -991,19 +996,22 @@ def __init__(
         self.impersonation_chain = impersonation_chain
         self.deferrable = deferrable
         self.polling_interval_seconds = polling_interval_seconds
+        self.ignore_if_missing = ignore_if_missing
 
     def execute(self, context: Context) -> None:
         hook = DataprocHook(gcp_conn_id=self.gcp_conn_id, impersonation_chain=self.impersonation_chain)
         try:
             op: operation.Operation = self._delete_cluster(hook)
 
         except NotFound:
-            self.log.info(
-                "Cluster %s not found in region %s. might have been deleted already.",
-                self.cluster_name,
-                self.region,
-            )
-            return
+            if self.ignore_if_missing:
+                self.log.info(
+                    "Cluster %s not found in region %s. Ignoring.",
+                    self.cluster_name,
+                    self.region,
+                )
+                return
+            raise
 
         except Exception as e:
             raise AirflowException(str(e))

providers/google/tests/unit/google/cloud/operators/test_dataproc.py

@@ -1289,6 +1289,7 @@ def test_execute_cluster_not_found(self, mock_hook):
             retry=RETRY,
             timeout=TIMEOUT,
             metadata=METADATA,
+            ignore_if_missing=True,
         )
 
         delete_cluster_op.execute(context=mock.MagicMock())
@@ -1304,9 +1305,7 @@ def test_execute_cluster_not_found(self, mock_hook):
         )
 
     @mock.patch(DATAPROC_PATH.format("DataprocHook"))
-    @mock.patch(DATAPROC_TRIGGERS_PATH.format("DataprocAsyncHook"))
-    def test_execute_cluster_not_found_deffered(self, mock_deffer, mock_hook):
-        mock_hook.return_value.create_cluster.return_value = None
+    def test_execute_cluster_not_found_raises_when_ignore_if_missing_false(self, mock_hook):
         mock_hook.return_value.delete_cluster.side_effect = NotFound("test")
         delete_cluster_op = DataprocDeleteClusterOperator(
             task_id="test_task",
@@ -1318,10 +1317,12 @@ def test_execute_cluster_not_found_deffered(self, mock_deffer, mock_hook):
             retry=RETRY,
             timeout=TIMEOUT,
             metadata=METADATA,
-            deferrable=True,
+            ignore_if_missing=False,
         )
 
-        delete_cluster_op.execute(context=mock.MagicMock())
+        with pytest.raises(NotFound):
+            delete_cluster_op.execute(context=mock.MagicMock())
+
         mock_hook.return_value.delete_cluster.assert_called_once_with(
             project_id=GCP_PROJECT,
             region=GCP_REGION,
@@ -1333,7 +1334,16 @@ def test_execute_cluster_not_found_deffered(self, mock_deffer, mock_hook):
             metadata=METADATA,
         )
 
-        assert not mock_deffer.called
+    def test_init_with_deferrable_and_ignore_if_missing_raises(self):
+        with pytest.raises(ValueError, match="Cannot set both deferrable and ignore_if_missing to True"):
+            DataprocDeleteClusterOperator(
+                task_id="test_task",
+                region=GCP_REGION,
+                cluster_name=CLUSTER_NAME,
+                project_id=GCP_PROJECT,
+                deferrable=True,
+                ignore_if_missing=True,
+            )
 
 
 class TestDataprocSubmitJobOperator(DataprocJobTestBase):