### Description

Google Cloud Dataproc cluster creation should eagerly delete ERROR state clusters.

It is possible for Google Cloud Dataproc clusters to create in the ERROR state.  The current operator (DataprocCreateClusterOperator) will require three total task attempts (original + two retries) in order to create the cluster, assuming underlying GCE infrastructure resolves itself between task attempts.  This can be reduced to two total attempts by eagerly deleting a cluster in ERROR state before failing the current task attempt. 

Clusters in the ERROR state are not useable to submit Dataproc based jobs via the Dataproc API. 



### Use case/motivation

Reducing the number of task attempts can reduce GCP based cost as delays between retry attempts could be minutes.  There's no reason to keep a running, costly cluster in the ERROR state if it can be detected in the initial create task. 

### Related issues

_No response_

### Are you willing to submit a PR?

- [X] Yes I am willing to submit a PR!

### Code of Conduct

- [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)

self.defer(
                trigger=DataprocWorkflowTrigger(
                    name=operation.operation.name,
                    project_id=self.project_id or hook.project_id,
                    region=self.region,
                    gcp_conn_id=self.gcp_conn_id,
                    impersonation_chain=self.impersonation_chain,
                    polling_interval_seconds=self.polling_interval_seconds,
                ),
                method_name="execute_complete",
            )
    def execute_complete(self, context, event=None) -> None:
        """Callback for when the trigger fires.
        This returns immediately. It relies on trigger to throw an exception,
        otherwise it assumes execution was successful.
        """
        if event["status"] == "failed" or event["status"] == "error":
            self.log.exception("Unexpected error in the operation.")
            raise AirflowException(event["message"])
        self.log.info("Workflow %s completed successfully", event["operation_name"])
class DataprocSubmitJobOperator(GoogleCloudBaseOperator):
    """Submit a job to a cluster.
    :param project_id: Optional. The ID of the Google Cloud project that the job belongs to.
    :param region: Required. The Cloud Dataproc region in which to handle the request.
    :param job: Required. The job resource.
        If a dict is provided, it must be of the same form as the protobuf message
        :class:`~google.cloud.dataproc_v1.types.Job`.
        For the complete list of supported job types and their configurations please take a look here
        https://cloud.google.com/dataproc/docs/reference/rest/v1/projects.regions.jobs
    :param request_id: Optional. A unique id used to identify the request. If the server receives two