Repository: apache/airflow
File: providers/google/src/airflow/providers/google/cloud/operators/dataproc.py (Python)

Code before:
        hook = DataprocHook(gcp_conn_id=self.gcp_conn_id, impersonation_chain=self.impersonation_chain)
        try:
            op: operation.Operation = self._delete_cluster(hook)
            if not self.deferrable:
                hook.wait_for_operation(timeout=self.timeout, result_retry=self.retry, operation=op)
                self.log.info("Cluster deleted.")
            else:
                end_time: float = time.time() + self.timeout
                self.defer(
                    trigger=DataprocDeleteClusterTrigger(
                        gcp_conn_id=self.gcp_conn_id,
                        project_id=self.project_id,
                        region=self.region,
                        cluster_name=self.cluster_name,
                        end_time=end_time,
                        metadata=self.metadata,
                        impersonation_chain=self.impersonation_chain,
                        polling_interval_seconds=self.polling_interval_seconds,
                    ),
                    method_name="execute_complete",
                )
        except NotFound:
            self.log.info(
                "Cluster %s not found in region %s. Skipping deletion.", self.cluster_name, self.region
            )
            raise AirflowSkipException(
                f"Cluster {self.cluster_name} in region {self.region} was not found - it may have already been deleted"
            )
        except Exception as e:
            raise AirflowException(str(e))

    def execute_complete(self, context: Context, event: dict[str, Any] | None = None) -> Any:
        """
        Act as a callback for when the trigger fires - returns immediately.

        Relies on trigger to throw an exception, otherwise it assumes execution was successful.
        """
        if event and event["status"] == "error":
            raise AirflowException(event["message"])
        if event is None:
            raise AirflowException("No event received in trigger callback")
        self.log.info("Cluster deleted.")

    def _delete_cluster(self, hook: DataprocHook):
        self.log.info("Deleting cluster: %s", self.cluster_name)
        return hook.delete_cluster(
            project_id=self.project_id,
            region=self.region,
            cluster_name=self.cluster_name,
            cluster_uuid=self.cluster_uuid,
            request_id=self.request_id,

Review comment (bug):
I am not sure why we need to add this exception? The task will fail anyway if the cluster can not be deleted, I am not sure that logic here should be like this with raising SkipException

Code after:
        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

        except Exception as e:
            raise AirflowException(str(e))

        if not self.deferrable:
            hook.wait_for_operation(timeout=self.timeout, result_retry=self.retry, operation=op)
            self.log.info("Cluster deleted.")
        else:
            end_time: float = time.time() + self.timeout
            self.defer(
                trigger=DataprocDeleteClusterTrigger(
                    gcp_conn_id=self.gcp_conn_id,
                    project_id=self.project_id,
                    region=self.region,
                    cluster_name=self.cluster_name,
                    end_time=end_time,
                    metadata=self.metadata,
                    impersonation_chain=self.impersonation_chain,
                    polling_interval_seconds=self.polling_interval_seconds,
                ),
                method_name="execute_complete",
            )

    def execute_complete(self, context: Context, event: dict[str, Any] | None = None) -> Any:
        """
        Act as a callback for when the trigger fires - returns immediately.

        Relies on trigger to throw an exception, otherwise it assumes execution was successful.
        """
        if event and event["status"] == "error":
            raise AirflowException(event["message"])
        if event is None:
            raise AirflowException("No event received in trigger callback")
        self.log.info("Cluster deleted.")

    def _delete_cluster(self, hook: DataprocHook):
        self.log.info("Deleting cluster: %s", self.cluster_name)
        return hook.delete_cluster(
            project_id=self.project_id,
            region=self.region,
            cluster_name=self.cluster_name,