[GitHub] [airflow] SamWheating commented on a diff in pull request #33668: Early delete a Dataproc cluster if started in the ERROR state.

["SamWheating (via GitHub)" <gi...@apache.org>]

SamWheating commented on code in PR #33668:
URL: https://github.com/apache/airflow/pull/33668#discussion_r1304924763


##########
airflow/providers/google/cloud/operators/dataproc.py:
##########
@@ -594,12 +594,16 @@ def _handle_error_state(self, hook: DataprocHook, cluster: Cluster) -> None:
         if cluster.status.state != cluster.status.State.ERROR:
             return
         self.log.info("Cluster is in ERROR state")
+        self.log.info("Gathering diagnostic information.")
         gcs_uri = hook.diagnose_cluster(
             region=self.region, cluster_name=self.cluster_name, project_id=self.project_id
         )
         self.log.info("Diagnostic information for cluster %s available at: %s", self.cluster_name, gcs_uri)
         if self.delete_on_error:
             self._delete_cluster(hook)
+            # The delete op is asynchronous and can cause further failure if the cluster finishes
+            # deleting between catching AlreadyExists and checking state
+            self._wait_for_cluster_in_deleting_state(hook)
             raise AirflowException("Cluster was created but was in ERROR state.")
         raise AirflowException("Cluster was created but is in ERROR state")

Review Comment:
   ```suggestion
           raise AirflowException("Cluster was created but is in ERROR state")
   ```
   
   I know its not a part of this PR, but can we remove this inner Exception? I think its redundant with the one below it, which would be raised anyways if this one were removed.



##########
airflow/providers/google/cloud/operators/dataproc.py:
##########
@@ -667,6 +671,23 @@ def execute(self, context: Context) -> dict:
                 raise
             self.log.info("Cluster already exists.")
             cluster = self._get_cluster(hook)
+        except AirflowException as ae:
+            # There still could be a cluster created here in an ERROR state which
+            # should be deleted immediately rather than consuming another retry attempt
+            # (assuming delete_on_error is true (default))
+            # This reduces overall the number of task attempts from 3 to 2 to successful cluster creation
+            # assuming the underlying GCE issues have resolved within that window. Users can configure
+            # a higher number of retry attempts in powers of two with 30s-60s wait interval
+            try:
+                cluster = self._get_cluster(hook)
+                # redundant condition checking in order to reuse _handle_error_sate
+                if cluster.status.state == cluster.status.State.ERROR:
+                    self._handle_error_state(hook, cluster)
+            except AirflowException:
+                # We could get any number of failures here, including cluster not found and we
+                # can just ignore to ensure we surface the original cluster create failure
+                pass

Review Comment:
   Can we log the exception rather than just swallowing it? There is some potentially useful information here such as a cluster being stuck in the deleting state which it might be worth communicating to users.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org