renu-j commented on code in PR #61951:
URL: https://github.com/apache/airflow/pull/61951#discussion_r2909339091
##########
providers/google/src/airflow/providers/google/cloud/operators/dataproc.py:
##########
@@ -805,11 +805,47 @@ def _retry_cluster_creation(self, hook: DataprocHook):
self.log.info("Cluster created.")
return Cluster.to_dict(cluster)
+ def _reconcile_cluster_state(self, hook: DataprocHook, cluster: Cluster)
-> Cluster:
+
+ if cluster.status.state == cluster.status.State.CREATING:
+ self.log.info("Cluster %s is in CREATING state.",
self.cluster_name)
+
+ cluster = self._wait_for_cluster_in_creating_state(hook)
+ self._handle_error_state(hook, cluster)
+ elif cluster.status.state == cluster.status.State.DELETING:
+ self.log.info("Cluster %s is in DELETING state.",
self.cluster_name)
+
+ self._wait_for_cluster_in_deleting_state(hook)
+
+ self.log.info("Attempting to re-create cluster: %s",
self.cluster_name)
+
+ operation = self._create_cluster(hook)
+ hook.wait_for_operation(
+ timeout=self.timeout,
+ result_retry=self.retry,
+ operation=operation,
+ )
+ cluster = self._get_cluster(hook)
+
Review Comment:
Thanks for the detailed explanation. That makes sense , I see the
distinction now between single-pass reconciliation and full state convergence.
Relying on the LRO completion as the signal for successful creation seems
consistent with the current operator contract. A stronger state validation
might indeed be better suited for a follow-up improvement rather than this
change.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
