Fix Cloud SQL 409 operationInProgress on import/export operations

[radhwene]

Problem

Cloud SQL allows only one administrative operation at a time per instance. When
CloudSQLImportInstanceOperator or CloudSQLExportInstanceOperator submits an
operation while another Cloud SQL admin operation is still running on the same
instance, the Cloud SQL Admin API returns HTTP 409 operationInProgress.

CloudSQLHook already uses GoogleBaseHook.operation_in_progress_retry() for
several Cloud SQL admin methods, but import_instance and export_instance
were not covered. As a result, transient backend contention fails the Airflow
task instead of retrying the operation submit.

Closes: #68040

Solution

This PR applies the existing operation_in_progress_retry() policy to:

• CloudSQLHook.import_instance
• CloudSQLHook.export_instance

It also fixes a latent exception-handling issue in import_instance.

Before this change, import_instance wrapped every HttpError into an
AirflowException. That prevented operation_in_progress_retry() from seeing
the original retryable HttpError.

With this change:

• retryable operationInProgress HttpErrors are re-raised unchanged so the
  retry decorator can evaluate them;
• terminal HttpErrors are still converted to the existing friendly
  AirflowException message.

There is no public API change, no new operator parameter, and no new
authentication surface.

Why hook-level retry, not only a sensor

A standalone "no operation in progress" sensor can be useful as an optional
pre-wait primitive, but it cannot fully fix this bug.

The sensor checks the Cloud SQL instance state before the import/export operator
submits the operation. There is still a race between:

1. the sensor observing the instance as idle;
2. the downstream task being scheduled;
3. the operator submitting the actual Cloud SQL Admin API request.

Another DAG, scheduler, user, maintenance operation, or external process can
start an admin operation during that gap. Therefore, the retry must exist at the
submit call itself.

This keeps the retry policy centralized in CloudSQLHook, where all operators
using these methods benefit automatically. It is also consistent with the
existing retry behavior already used by other Cloud SQL admin methods.

A sensor can still be added separately as an optional convenience, but the
correctness fix for 409 operationInProgress belongs in the hook.

Testing

This PR is covered at two levels.

Unit tests

The unit tests verify the local retry contract:

• export_instance retries retryable HttpError responses.
• import_instance retries retryable HttpError responses.
• import_instance re-raises retryable operationInProgress HttpError
  unchanged, so operation_in_progress_retry() can evaluate the original
  exception type.
• terminal HttpErrors are still converted to the existing AirflowException
  message.

The assertions are based on exception types and retry behavior, not string
matching on error messages.

E2E validation

The fix was also validated against a real Cloud SQL PostgreSQL instance.

• Stock provider: parallel Cloud SQL admin operations against the same instance
  reproduce 409 operationInProgress.
• Patched provider: the same topology succeeds after retrying the submit.
• The E2E scenario was run twice to reduce flakiness risk.

The E2E DAG submits imports and exports in parallel against the same instance,
which validates both patched hook methods under real Cloud SQL operation
serialization.

Breaking changes

None.

The public API is unchanged. The only behavioral change is that Cloud SQL
import and export now retry the same transient backend contention condition
already handled by other Cloud SQL admin methods.

---

Important

🛠️ Maintainer triage note for @radhwene · by @potiuk · 2026-06-17 14:51 UTC

Helpful heads-up from the maintainers — please address before this PR can be reviewed:

• Static / docs checks failing (CI image checks / Static checks). Run them locally with prek run --all-files (or pre-commit run --all-files) and push the fixes.
• Failing test jobs: Non-DB tests: providers / Non-DB-prov::3.10:-amazon,celer...standard, Special tests / Latest Boto test: providers / All-prov:LatestBoto-Postgres:14:3.10:-amazon,celer...standard, Special tests / Pendulum2 test: providers / All-prov:Pendulum2-Postgres:14:3.10:-amazon,celer...standard, provider distributions tests / Compat 2.11.1:P3.10: … (+3 more). Reproduce and fix locally, then push.
• See the Pull Request quality criteria.

The ball is in your court — you've been assigned to this PR. Fix the above, then mark it Ready for review.

_{Automated triage — may be imperfect; a maintainer takes the next look.}

[radhwene]

E2E evidence: real Cloud SQL PostgreSQL instance, Airflow v2

Adding the reproduction and validation runs behind the “Why hook-level retry, not only a sensor”
and “E2E validation” sections.

Sensor-only approach does not close the race.

I validated the standalone CloudSQLNoOperationInProgressSensor approach by placing a
CloudSQLNoOperationInProgressSensor in reschedule mode before each import task.

In that run, the sensor task wait_a reports success, but the immediately downstream
CloudSQLImportInstanceOperator task import_complaints still fails with
409 operationInProgress.

That demonstrates the remaining TOCTOU window between:

1. the sensor observing the instance as idle;
2. the downstream operator being scheduled;
3. the operator submitting the actual Cloud SQL Admin API request.

Hook-level retry handles the contention at the submit point.

I also validated this PR with a harder topology: four Cloud SQL admin operations
submitted in parallel against the same instance — two imports and two exports — with no
sensor.

All four tasks succeed after retrying the operation submit, and no 409 operationInProgress
is surfaced to the DAG run.

This validates both patched hook methods, import_instance and export_instance, under
real Cloud SQL operation serialization.

E2E DAG used for the validation (cloudsql_retry_stress_409)

"""
cloudsql_retry_stress.py

E2E stress test for the apache/airflow#68040 retry-on-409 fix.

This DAG fans out four Cloud SQL admin operations in parallel against the same
Cloud SQL PostgreSQL instance:

    create_tables ──┬──> import_complaints
                    ├──> import_crime
                    ├──> export_a
                    └──> export_b

Cloud SQL serializes admin operations per instance, so simultaneous submits
can collide with HTTP 409 ``operationInProgress``.

Expected behavior:

  - stock provider:
      parallel operations can fail with 409 and the DAG run fails;

  - patched provider:
      import/export submit calls retry through ``operation_in_progress_retry``;
      all four tasks complete successfully.

Export object URIs are unique per run with ``{{ ts_nodash }}`` because Cloud SQL
export fails if the destination GCS object already exists.

Trigger config: PROJECT_ID, INSTANCE, GCS_BUCKET, DB_NAME.
"""

from __future__ import annotations

from datetime import datetime
from pathlib import Path

from airflow import DAG
from airflow.models.param import Param
from airflow.providers.google.cloud.operators.cloud_sql import (
    CloudSQLExportInstanceOperator,
    CloudSQLImportInstanceOperator,
)
from airflow.providers.postgres.operators.postgres import PostgresOperator

INIT_SQL_PATH = Path(__file__).resolve().parent / "init_tables.sql"


def _import(task_id: str, table: str) -> CloudSQLImportInstanceOperator:
    return CloudSQLImportInstanceOperator(
        task_id=task_id,
        project_id="{{ params.PROJECT_ID }}",
        instance="{{ params.INSTANCE }}",
        body={
            "importContext": {
                "fileType": "CSV",
                "uri": f"gs://{{{{ params.GCS_BUCKET }}}}/{table}.csv",
                "database": "{{ params.DB_NAME }}",
                "csvImportOptions": {"table": table},
            }
        },
        gcp_conn_id="google_cloud_default",
    )


def _export(task_id: str, label: str) -> CloudSQLExportInstanceOperator:
    return CloudSQLExportInstanceOperator(
        task_id=task_id,
        project_id="{{ params.PROJECT_ID }}",
        instance="{{ params.INSTANCE }}",
        body={
            "exportContext": {
                "fileType": "CSV",
                "uri": f"gs://{{{{ params.GCS_BUCKET }}}}/stress_export_{label}_{{{{ ts_nodash }}}}.csv",
                "databases": ["{{ params.DB_NAME }}"],
                "csvExportOptions": {"selectQuery": "SELECT 1 AS col"},
            }
        },
        gcp_conn_id="google_cloud_default",
    )


with DAG(
    dag_id="cloudsql_retry_stress_409",
    start_date=datetime(2024, 1, 1),
    schedule=None,
    catchup=False,
    tags=["cloudsql", "fix-409", "retry", "stress"],
    params={
        "PROJECT_ID": Param(default="CHANGE_ME", type="string"),
        "INSTANCE": Param(default="CHANGE_ME", type="string"),
        "GCS_BUCKET": Param(default="CHANGE_ME", type="string"),
        "DB_NAME": Param(default="airflow_db", type="string"),
    },
    render_template_as_native_obj=True,
) as dag:

    create_tables = PostgresOperator(
        task_id="create_tables",
        postgres_conn_id="cloudsql_pg",
        sql=INIT_SQL_PATH.read_text(),
    )

    create_tables >> [
        _import("import_complaints", "complaints"),
        _import("import_crime", "crime"),
        _export("export_a", "a"),
        _export("export_b", "b"),
    ]

[henry3260]

Thanks for the fix!

[radhwene]

Rebased onto current main and force-pushed commit Id 0e73e09 .

Hi @potiuk The earlier red CI was not from this change:
the branch had picked up a Merge branch 'main' commit that CI evaluated against a 2-day-stale main. The static-check failure was a merge artifact (the new-AirflowException allowlist matches the 33 real raises, and mypy-providers passed), and the failing provider matrix jobs (Non-DB, LatestBoto, Pendulum2, Compat) were the azure-storage-blob==12.30.0 breakage (#68482) — this diff touches no Azure files. The rebase drops the merge commit and pulls in the merged Azure fix, so the provider jobs run clean now.

@henry3260 's review feedback is addressed: the operation_in_progress_retry decorator is now scoped to the import submit only (_submit_import), with the wait kept outside the retry boundary so a polling 429 can never
re-submit the import. Regression test added.

[radhwene]

Rebased on current main today. No conflicts, GitHub reports it as mergeable.

The changes requested on 2026-06-12 got fixed the next day. operation_in_progress_retry()
used to wrap all of import_instance, polling included, so a retryable error while waiting
re-ran the whole method and submitted the import a second time. The submit now sits in
_submit_import, and only that call carries the decorator. Polling runs outside the retry
scope, so if it fails the task fails instead of importing the same data twice. There's a
regression test asserting exactly one submit when polling raises a retryable error.

The changes-requested review is still open seven weeks later, and it's the only thing
holding this back. Could someone take another look? I'll rebase again if it goes stale.