Revert "Add 'ignore_if_missing' to DataprocDeleteClusterOperator (#60433)

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

@@ -949,8 +949,6 @@ 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] = (
@@ -976,7 +974,6 @@ 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)
@@ -994,22 +991,19 @@ 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:
-            if self.ignore_if_missing:
-                self.log.info(
-                    "Cluster %s not found in region %s. Ignoring.",
-                    self.cluster_name,
-                    self.region,
-                )
-                return
-            raise
+            self.log.info(
+                "Cluster %s not found in region %s. might have been deleted already.",
+                self.cluster_name,
+                self.region,
+            )
+            return
 
         except Exception as e:
             raise AirflowException(str(e))

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

@@ -1289,7 +1289,6 @@ 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())
@@ -1320,7 +1319,6 @@ def test_execute_cluster_not_found_deffered(self, mock_deffer, mock_hook):
             timeout=TIMEOUT,
             metadata=METADATA,
             deferrable=True,
-            ignore_if_missing=True,
         )
 
         delete_cluster_op.execute(context=mock.MagicMock())
@@ -1337,36 +1335,6 @@ def test_execute_cluster_not_found_deffered(self, mock_deffer, mock_hook):
 
         assert not mock_deffer.called
 
-    @mock.patch(DATAPROC_PATH.format("DataprocHook"))
-    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",
-            region=GCP_REGION,
-            cluster_name=CLUSTER_NAME,
-            project_id=GCP_PROJECT,
-            cluster_uuid=None,
-            request_id=REQUEST_ID,
-            retry=RETRY,
-            timeout=TIMEOUT,
-            metadata=METADATA,
-            ignore_if_missing=False,
-        )
-
-        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,
-            cluster_name=CLUSTER_NAME,
-            cluster_uuid=None,
-            request_id=REQUEST_ID,
-            retry=RETRY,
-            timeout=TIMEOUT,
-            metadata=METADATA,
-        )
-
 
 class TestDataprocSubmitJobOperator(DataprocJobTestBase):
     @mock.patch(DATAPROC_PATH.format("DataprocHook"))