omkar-foss commented on code in PR #68361:
URL: https://github.com/apache/airflow/pull/68361#discussion_r3728225183
##########
providers/google/src/airflow/providers/google/cloud/hooks/cloud_sql.py:
##########
@@ -354,11 +364,54 @@ def export_instance(self, instance: str, body: dict,
project_id: str):
operation_name = response["name"]
return operation_name
+ @GoogleBaseHook.operation_in_progress_retry()
+ def _submit_import(self, instance: str, body: dict, project_id: str) ->
str:
+ """
+ Submit an import request for a Cloud SQL instance, retrying while an
operation is in progress.
+
+ Cloud SQL serializes administrative operations per instance, so
submitting an import while
+ another admin operation is already running on the same instance fails
with HTTP 409
+ ``operationInProgress``. The ``operation_in_progress_retry`` decorator
retries the submit on
+ HTTP 409/429 with exponential backoff (capped at 300s between
attempts) until the API accepts
+ it. The decorator deliberately wraps only this submit call: a 409/429
here guarantees the
+ import was rejected, so repeating it is safe.
+
+ :param instance: Database instance ID. This does not include the
project ID.
+ :param body: The request body, as described in
+
https://cloud.google.com/sql/docs/mysql/admin-api/v1beta4/instances/import#request-body
+ :param project_id: Project ID of the project that contains the
instance.
+ :return: The name of the accepted import operation.
+ """
+ try:
+ response = (
+ self.get_conn()
+ .instances()
+ .import_(project=project_id, instance=instance, body=body)
+ .execute(num_retries=self.num_retries)
+ )
+ return response["name"]
+ except HttpError as ex:
+ # ``operation_in_progress_retry`` retries on the raw ``HttpError``
(status 409/429), not on
+ # ``AirflowException``. Re-raise operation-in-progress errors
unchanged so the decorator can
+ # see and retry them; otherwise the 409 would be wrapped below and
the retry never triggers.
+ # Genuinely terminal HttpErrors still get the friendly message.
+ if is_operation_in_progress_exception(ex):
+ raise
+ raise AirflowException(f"Importing instance {instance} failed:
{ex.content}")
+
@GoogleBaseHook.fallback_to_default_project_id
def import_instance(self, instance: str, body: dict, project_id: str) ->
None:
"""
Import data into a Cloud SQL instance from a SQL dump or CSV file in
Cloud Storage.
+ The submit call is retried on HTTP 409 ``operationInProgress`` / 429
(see ``_submit_import``),
Review Comment:
You could make some of the comments in your diff a bit more concise, if
you'd like.
--
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]
