Cloud SQL — CloudSQLImportInstanceOperator / CloudSQLExportInstanceOperator 409 operationInProgress on parallel tasks against the same instanc

Providers

### Apache Airflow version

2.10.3 or 3.2.0 +

### What happened and how to reproduce it?

# Cloud SQL — `CloudSQLImportInstanceOperator` / `CloudSQLExportInstanceOperator` 409 `operationInProgress` on parallel tasks against the same instance

## Problem

Two `CloudSQLImportInstanceOperator` tasks against the same instance, no dependency between them → fan out in parallel → one succeeds, the other fails in <2s with HTTP 409 / `operationInProgress`. Same race on `CloudSQLExportInstanceOperator` and on IMPORT+EXPORT mixed: Cloud SQL serializes admin ops per instance. Provider does not retry. No sensor in the provider waits for the instance to be idle.

## Versions

| Component | Version |
| --- | --- |
| Apache Airflow | 2.10.3 |
| apache-airflow-providers-google | 10.22.0 |
| apache-airflow-providers-postgres | 5.11.1 |
| Cloud SQL instance | `my-cloudsql-pg` (PostgreSQL 16, `us-central1`) |
| Executor | LocalExecutor |
| Python | 3.12.7 |

Also reproduces on Airflow 3.2.0 + providers-google 19.5.0.

## Reproduction DAG

```python
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 (
CloudSQLImportInstanceOperator,
)
from airflow.providers.postgres.operators.postgres import PostgresOperator

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

with DAG(
dag_id="reproduce_409_cloudsql_import",
start_date=datetime(2024, 1, 1),
schedule=None,
catchup=False,
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(),
)

import_complaints = CloudSQLImportInstanceOperator(
task_id="import_complaints",
project_id="{{ params.PROJECT_ID }}",
instance="{{ params.INSTANCE }}",
body={"importContext": {
"fileType": "CSV",
"uri": "gs://{{ params.GCS_BUCKET }}/complaints.csv",
"database": "{{ params.DB_NAME }}",
"csvImportOptions": {"table": "complaints"},
}},
gcp_conn_id="google_cloud_default",
)

import_crime = CloudSQLImportInstanceOperator(
task_id="import_crime",
project_id="{{ params.PROJECT_ID }}",
instance="{{ params.INSTANCE }}",
body={"importContext": {
"fileType": "CSV",
"uri": "gs://{{ params.GCS_BUCKET }}/crime.csv",
"database": "{{ params.DB_NAME }}",
"csvImportOptions": {"table": "crime"},
}},
gcp_conn_id="google_cloud_default",
)

create_tables >> [import_complaints, import_crime]
```

Triggered with:

```
airflow dags trigger reproduce_409_cloudsql_import \
--run-id repro_409_20260604_215213 \
--conf '{"PROJECT_ID":"my-project","INSTANCE":"my-cloudsql-pg","GCS_BUCKET":"my-gcs-bucket","DB_NAME":"airflow_db"}'
```

## Result — `import_complaints` failed with HTTP 409 `operationInProgress`


## Result — `import_crime` succeeded (same DAG, same instance, parallel task)


## `gcloud sql operations list` — only one IMPORT op was created

```
gcloud sql operations list \
--instance=my-cloudsql-pg \
--project=my-project
```


Single IMPORT op at `2026-06-04T19:52:32` (the successful `import_crime`). The failed `import_complaints` produced no entry — the API rejected the request before allocating a long-running operation. Documented behavior: [one administrative operation per instance at a time][cloudsql-ops].

[cloudsql-ops]: https://docs.cloud.google.com/sql/docs/postgres/admin-api#long-running

## Current workarounds (all unsatisfactory)

| Workaround | Why it falls short |
| --- | --- |
| Serialize tasks (`a >> b` instead of `[a, b]`) | Kills parallelism on purpose. Doesn't scale to N imports/exports. Doesn't protect against a second DAG hitting the same instance. |
| Airflow `Pool(size=1)` per Cloud SQL instance | Coarse-grained: blocks all tasks of any operator type in that pool, not just admin ops. Requires one pool per instance, declared out-of-band. Still races against console-triggered ops, backups, replica events. |
| Task-level `retries=N` + `retry_delay` | Blunt: doesn't know if the op actually started server-side. On import, a blind retry can either replay rows already landed or hit the same 409 indefinitely. Wastes worker slots holding tasks queued for retry. |
| Custom `PythonOperator` pre-check polling `sqladmin.operations.list` | Works but duplicates logic that arguably belongs in the provider. Every team writes it slightly differently. No async support → blocks a worker slot for the whole wait. |

## Proposal — deferrable instance-level sensor

```python
wait_for_slot = CloudSQLNoOperationInProgressSensor(
task_id="wait_for_slot",
project_id="my-project",
instance="my-cloudsql-pg",
poke_interval=60,
timeout=60 * 60 * 2,
deferrable=True,
)

wait_for_slot >> CloudSQLImportInstanceOperator(...)
```

- Polls `sqladmin.operations.list`, filters on `targetId == instance` and on any op in a non-terminal state.
- Async trigger reuses `CloudSQLHook`. No new auth surface.
- Fails fast on 404 (instance missing) / 403 (IAM denied).
- Operation-agnostic → covers IMPORT, EXPORT, and any future `sqladmin.instances.*` operator.

## Contribution plan

One PR — sensor + async trigger + unit tests + short docs note. Will open as soon as direction is confirmed.

### What you think should happen instead?

The Google provider should ship a deferrable instance-level sensor (e.g. `CloudSQLNoOperationInProgressSensor`) that waits until the target Cloud SQL instance has no
non-terminal administrative operation in flight, then releases.

With this sensor wired upstream of `CloudSQLImportInstanceOperator` / `CloudSQLExportInstanceOperator`, parallel admin ops against the same instance no longer fail with HTTP 409
`operationInProgress`. The operator's submit semantics stay unchanged; the sensor is the explicit "wait for slot" primitive that every team currently re-implements with a
custom `PythonOperator` polling `sqladmin.operations.list`.

Full design + workaround comparison in the issue body.

### Operating System

_No response_

### Deployment

None

### Apache Airflow Provider(s)

google

### Versions of Apache Airflow Providers

_No response_

### Official Helm Chart version

Not Applicable

### Kubernetes Version

_No response_

### Helm Chart configuration

_No response_

### Docker Image customizations

_No response_

### Anything else?

_No response_

### Are you willing to submit 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)