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

Code before:
        impersonation_chain: str | Sequence[str] | None = None,
        deferrable: bool = conf.getboolean("operators", "default_deferrable", fallback=False),
        polling_interval_seconds: int = 10,
        **kwargs,
    ):
        super().__init__(**kwargs)
        if deferrable and polling_interval_seconds <= 0:
            raise ValueError("Invalid value for polling_interval_seconds. Expected value greater than 0")
        self.project_id = project_id
        self.region = region
        self.cluster_name = cluster_name
        self.cluster_uuid = cluster_uuid
        self.request_id = request_id
        self.retry = retry
        self.timeout = timeout
        self.metadata = metadata
        self.gcp_conn_id = gcp_conn_id
        self.impersonation_chain = impersonation_chain
        self.deferrable = deferrable
        self.polling_interval_seconds = polling_interval_seconds

    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)
            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:

Review comment (suggestion):
NotFound exception I think will not be raised here from deferrable mode. If it will be raised, it will be raised by the API call, not the def mode itself.

Code after:
        deferrable: bool = conf.getboolean("operators", "default_deferrable", fallback=False),
        polling_interval_seconds: int = 10,
        **kwargs,
    ):
        super().__init__(**kwargs)
        if deferrable and polling_interval_seconds <= 0:
            raise ValueError("Invalid value for polling_interval_seconds. Expected value greater than 0")
        self.project_id = project_id
        self.region = region
        self.cluster_name = cluster_name
        self.cluster_uuid = cluster_uuid
        self.request_id = request_id
        self.retry = retry
        self.timeout = timeout
        self.metadata = metadata
        self.gcp_conn_id = gcp_conn_id
        self.impersonation_chain = impersonation_chain
        self.deferrable = deferrable
        self.polling_interval_seconds = polling_interval_seconds

    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

        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,