Introducing object store backend for task and asset store

[amoghrajesh]

---

Was generative AI tooling used to co-author this PR?

• Yes: claude sonnet 4.6

Built atop: #68274 (only last commit is relevant)

What

AIP-103 introduced task and asset state management backed by the metadata database. For deployments that need to store large state values as a custom backend, this PR adds an object storage alternative via apache-airflow-providers-common-io, similar to https://airflow.apache.org/docs/apache-airflow-providers-common-io/stable/xcom_backend.html

Current behaviour

Task and asset state can only be stored in the Airflow metadata database (MetastoreStateBackend). There is a way to offload values to S3, GCS, Azure Blob, or any other object storage, but there is no such example backend to do so.

Proposed change

Adds StoreObjectStorageBackend to providers/common/io, which stores task and asset state on object storage using ObjectStoragePath. The backend supports:

• Threshold-based offloading: store_objectstorage_threshold = 0 (default) offloads all values to the backend, set a positive byte count to keep small values inline in the database and only offload large ones.
• Optional compression: set store_objectstorage_compression = gzip (or any fsspec-supported codec).
• Stable paths across retries: task state is keyed on (dag_id, run_id, task_id, map_index), making this backend suitable for operators using ResumableJobMixin.

Enable it by setting in airflow.cfg (or env vars using COMMON_IO as the section, e.g. AIRFLOW__COMMON_IO__STATE_STORE_OBJECTSTORAGE_PATH):

[state_store]
backend = airflow.providers.common.io.store.backend.StoreObjectStorageBackend

[common.io]
state_store_objectstorage_path = s3://conn_id@mybucket/task-state/
statestore_objectstorage_threshold = 0
state_store_objectstorage_compression = gzip  # optional

Changes of Note

Values persist until explicitly deleted. Use your object storage providers lifecycle policies (S3 lifecycle rules, GCS object lifecycle, etc.) to expire old state automatically.

The backend requires airflow >= 3.3 (when BaseStoreBackend and the AssetScope/TaskScope types were introduced).

User implications / backcompat

New opt-in feature, no changes to existing deployments. Requires adding a connection in Airflow with the object storage credentials (e.g. endpoint_url for minio).

Testing

Verified end-to-end using the example_task_store example Dag with minio running locally. Three scenarios were tested:

Setup

1. Start MinIO: docker run -p 29000:9000 -p 29001:9001 -e MINIO_ROOT_USER=minioadmin -e MINIO_ROOT_PASSWORD=minioadmin quay.io/minio/minio server /data --console-address ":9001"
2. Create bucket airflow-task-state via minio console (http://localhost:29001).

3. Add an Airflow connection minio with conn_type=aws, login=minioadmin, password=minioadmin, extra={"endpoint_url": "http://host.docker.internal:29000"} (use host.docker.internal when running inside Breeze/Docker).
4. Set the base env vars and restart Airflow so and workers inherit them:

export AIRFLOW__WORKERS__STATE_STORE_BACKEND=airflow.providers.common.io.store.backend.StoreObjectStorageBackend
export AIRFLOW__COMMON_IO__STATE_STORE_OBJECTSTORAGE_PATH=s3://minio@airflow-task-state/task-state/
export AIRFLOW__COMMON_IO__STATE_STORE_OBJECTSTORAGE_THRESHOLD=0

export AIRFLOW_CONN_MINIO='{"conn_type": "aws", "login": "minioadmin", "password": "minioadmin", "extra": {"endpoint_url": "http://host.docker.internal:29000"}}'

Scenario 1 — All values offloaded (threshold=0)

AIRFLOW__COMMON_IO__STATE_STORE_OBJECTSTORAGE_THRESHOLD=0

Verified: files appear in MinIO under task-state/example_task_store/.../. On try 1, the job ID is written and the task intentionally fails. On try 2, the job ID is read back from MinIO and the task reattaches to the existing job.

Try 1:

Try 2:

Task store tab:

Minio:

---

Scenario 2 — Threshold: small values stay in DB, large ones go to MinIO

AIRFLOW__COMMON_IO__STATE_STORE_OBJECTSTORAGE_THRESHOLD=50

Result:

Out of the 4 task store values, only result is greater than 50bytes and is stored in the minio backend. Rest in database

Try 1:

Try 2:

---

Scenario 3 — Compression with gzip

AIRFLOW__COMMON_IO__STATE_STORE_OBJECTSTORAGE_THRESHOLD=0
AIRFLOW__COMMON_IO__STATE_STORE_OBJECTSTORAGE_COMPRESSION=gzip

Verified: files appear in MinIO with a .gz suffix (e.g. task-state/example_task_store/.../job_id.gz). Decompression is inferred automatically on read (compression="infer"). Task completes successfully.

Try 1:

Try 2:

Minio:

~/D/O/r/airflow ❯❯❯ cd ~/Downloads                                                                                                                                                aip
~/Downloads ❯❯❯ open result.gz

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[ashb]

[common.io]
store_objectstorage_path = s3://conn_id@mybucket/task-state/

This feels subtly wrong. Why isn't this "just" using the existing configuration for the Object storage?

[amoghrajesh]

@ashb I can imagine a case when someone ONLY wants to use custom backend for task store + asset store and not for xcoms. Both should be independently configurable I think

[kaxil]

Let's actually rethink this, since on the other PR we are saying we want to limit "State Store" to certain MB anyway.

Going to defer this to after 3.3

[ianbuss]

Optional clarification in doc

[amoghrajesh]

Needs a little more update due to #68438

Outdated

[amoghrajesh]

Tested it end to end again, works as expected.

[uranusjr]

From what I can tell, the config in main is [workers] state_store_backend instead of [state_store] backend. Is this something you intend to change in a differnt PR? I didn’t find mentions to this discrepency.

[amoghrajesh]

From what I can tell, the config in main is [workers] state_store_backend instead of [state_store] backend. Is this something you intend to change in a differnt PR? I didn’t find mentions to this discrepency.

Good catch, both sections exist but serve different roles. [state_store] backend is the server-side backend (controls how the API server persists state to the metadata DB). [workers] state_store_backend is the worker-side offload backend that StateStoreObjectStorageBackend plugs into and workers use it to write large values directly to object storage and store only a reference in the DB. The docs have the wrong section; fixed in 0461d81

[Lee-W]

Mostly nitpicks on tests and private methods

[amoghrajesh]

Mostly nitpicks on tests and private methods

Handled all in comments from wei

[kaxil]

Left 1 comment, feel free to merge after addressing