You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by ra...@apache.org on 2026/07/03 10:56:37 UTC
(airflow) branch v3-3-test updated: [v3-3-test] Rename `state-store cleanup-task-state-store` command to `state-store clean` (#69316) (#69322)
This is an automated email from the ASF dual-hosted git repository.
vatsrahul1001 pushed a commit to branch v3-3-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-3-test by this push:
new d7fb68d6c1a [v3-3-test] Rename `state-store cleanup-task-state-store` command to `state-store clean` (#69316) (#69322)
d7fb68d6c1a is described below
commit d7fb68d6c1a858d5280abfb889f0bc17c8955642
Author: github-actions[bot] <41...@users.noreply.github.com>
AuthorDate: Fri Jul 3 16:26:27 2026 +0530
[v3-3-test] Rename `state-store cleanup-task-state-store` command to `state-store clean` (#69316) (#69322)
The command added in 3.3.0rc1 repeated "state store" in both the group and the
subcommand. Rename it to `airflow state-store clean`, mirroring `airflow db
clean`. The old name only shipped in 3.3.0rc1, so renaming before GA keeps a
single stable name for 3.3.0.
Also make the help text and docs accurate about the current behavior: the
command intentionally supports the metastore backend only and skips custom
(worker-side) backends, and the docstring now explains why (correct cleanup of a
worker-side backend needs to delete the DB refs and the backend data in order,
which has to run where the backend's dependencies live, not on the server-side
CLI). Fix a wrong class name (`MetastoreStateBackend` -> `MetastoreBackend`) in
the docs.
(cherry picked from commit 71be2b47b339bf7c76b3cb0b84f50261f3d830e2)
Co-authored-by: Kaxil Naik <ka...@gmail.com>
---
.../task-and-asset-state-store-cleanup.rst | 4 +--
.../task-and-asset-state-store.rst | 6 ++--
airflow-core/src/airflow/cli/cli_config.py | 13 +++----
.../airflow/cli/commands/state_store_command.py | 17 +++++++--
.../unit/cli/commands/test_state_store_command.py | 40 ++++++++++++++--------
5 files changed, 52 insertions(+), 28 deletions(-)
diff --git a/airflow-core/docs/administration-and-deployment/task-and-asset-state-store-cleanup.rst b/airflow-core/docs/administration-and-deployment/task-and-asset-state-store-cleanup.rst
index b20a8325767..58a5073f189 100644
--- a/airflow-core/docs/administration-and-deployment/task-and-asset-state-store-cleanup.rst
+++ b/airflow-core/docs/administration-and-deployment/task-and-asset-state-store-cleanup.rst
@@ -48,7 +48,7 @@ Running cleanup
The command is::
- airflow state-store cleanup-task-state-store
+ airflow state-store clean
It reads ``[state_store] default_retention_days`` and ``[state_store] state_cleanup_batch_size`` from the ``airflow.cfg`` file, then deletes all eligible rows.
@@ -56,7 +56,7 @@ It reads ``[state_store] default_retention_days`` and ``[state_store] state_clea
Use ``--dry-run`` to preview what would be deleted without removing anything::
- airflow state-store cleanup-task-state-store --dry-run
+ airflow state-store clean --dry-run
The output lists every row that would be deleted, grouped by dag, run, task, map index, and key.
diff --git a/airflow-core/docs/administration-and-deployment/task-and-asset-state-store.rst b/airflow-core/docs/administration-and-deployment/task-and-asset-state-store.rst
index e3239346450..5ae6c0b1b33 100644
--- a/airflow-core/docs/administration-and-deployment/task-and-asset-state-store.rst
+++ b/airflow-core/docs/administration-and-deployment/task-and-asset-state-store.rst
@@ -99,7 +99,7 @@ When this is set, ``TaskStateStoreAccessor.set()`` calls ``serialize_task_state_
Garbage collection semantics
-----------------------------
-The cleanup task, also known as "garbage collection" is triggered using the Airflow CLI. The command to trigger the cleanup task is ``airflow state-store cleanup-task-state-store``. This process removes store rows according to the following rules:
+The cleanup task, also known as "garbage collection" is triggered using the Airflow CLI. The command to trigger the cleanup task is ``airflow state-store clean``. This process removes store rows according to the following rules:
**Time-based expiry (task state store only)**
Rows whose ``expires_at < now()`` are deleted. ``expires_at`` is computed on the *worker* at write time, not by the server.
@@ -115,7 +115,7 @@ The cleanup task, also known as "garbage collection" is triggered using the Airf
.. important::
- Garbage collection only works for the ``MetastoreStateBackend``. Custom backends are explicitly skipped.
+ Garbage collection only works for the ``MetastoreBackend``. Custom backends are explicitly skipped.
@@ -207,5 +207,5 @@ Example skeleton:
return json.loads(s3_object["Body"].read().decode())
# Implement the remaining abstract methods as pass-throughs or delegating to the
- # default MetastoreStateBackend for the DB side
+ # default MetastoreBackend for the DB side
...
diff --git a/airflow-core/src/airflow/cli/cli_config.py b/airflow-core/src/airflow/cli/cli_config.py
index 44a9fd47c23..46a0ebfa391 100644
--- a/airflow-core/src/airflow/cli/cli_config.py
+++ b/airflow-core/src/airflow/cli/cli_config.py
@@ -1684,14 +1684,15 @@ TEAMS_COMMANDS = (
)
STATE_STORE_COMMANDS = (
ActionCommand(
- name="cleanup-task-state-store",
- help="Remove expired task state store rows (MetastoreBackend only)",
+ name="clean",
+ help="Remove expired task state store rows (metastore backend only)",
description=(
- "Reads [state_store] default_retention_days from config and deletes task_state_store rows "
- "older than the configured threshold. Only applies when MetastoreBackend is configured; "
- "custom backends are skipped. Use --dry-run to preview without deleting."
+ "Deletes task_state_store rows whose expires_at is in the past, honoring the state_store "
+ "settings default_retention_days and state_cleanup_batch_size. Currently supports the "
+ "default metastore backend only; custom (worker-side) backends are skipped. Use --dry-run "
+ "to preview deletions."
),
- func=lazy_load_command("airflow.cli.commands.state_store_command.cleanup_task_state_store"),
+ func=lazy_load_command("airflow.cli.commands.state_store_command.clean_state_store"),
args=(ARG_DB_DRY_RUN, ARG_VERBOSE),
),
)
diff --git a/airflow-core/src/airflow/cli/commands/state_store_command.py b/airflow-core/src/airflow/cli/commands/state_store_command.py
index 652c40d3445..c5139522d61 100644
--- a/airflow-core/src/airflow/cli/commands/state_store_command.py
+++ b/airflow-core/src/airflow/cli/commands/state_store_command.py
@@ -26,12 +26,23 @@ log = logging.getLogger(__name__)
# Other state operations (list, get, delete per key) will be added here in the future.
-def cleanup_task_state_store(args) -> None:
- """Remove expired task state store rows (MetastoreBackend only)."""
+def clean_state_store(args) -> None:
+ """
+ Remove expired task state store rows from the metastore backend.
+
+ Deliberately restricted to ``MetastoreBackend`` for now. A custom backend is typically
+ worker-side (object storage, etc.); cleaning it correctly means deleting the metadata-DB
+ refs *and* the backend data in order, which has to run where the backend and its
+ dependencies live (the worker), not on the server-side CLI. Until that experience exists,
+ the command skips custom backends rather than half-cleaning them.
+ """
backend = get_state_backend()
if not isinstance(backend, MetastoreBackend):
- print("Custom backend configured — skipping cleanup (not supported).")
+ print(
+ f"Custom state store backend configured ({type(backend).__name__}); skipping. "
+ f"The clean command currently supports the metastore backend only."
+ )
return
if args.dry_run:
diff --git a/airflow-core/tests/unit/cli/commands/test_state_store_command.py b/airflow-core/tests/unit/cli/commands/test_state_store_command.py
index 652e08fe110..d777d41afd7 100644
--- a/airflow-core/tests/unit/cli/commands/test_state_store_command.py
+++ b/airflow-core/tests/unit/cli/commands/test_state_store_command.py
@@ -22,21 +22,26 @@ from unittest.mock import MagicMock, patch
import pytest
-from airflow.cli.commands.state_store_command import cleanup_task_state_store
+from airflow._shared.state import BaseStoreBackend
+from airflow.cli.commands.state_store_command import clean_state_store
from airflow.state.metastore import MetastoreBackend
pytestmark = pytest.mark.db_test
-class TestStateStoreCleanupCommand:
- def test_cleanup_calls_backend(self):
+class TestStateStoreCleanCommand:
+ def test_clean_calls_backend(self):
args = Namespace(dry_run=False, verbose=False)
backend = MetastoreBackend()
with (
- mock.patch("airflow.cli.commands.state_store_command.get_state_backend", return_value=backend),
- patch.object(backend, "cleanup"),
+ mock.patch(
+ "airflow.cli.commands.state_store_command.get_state_backend",
+ return_value=backend,
+ autospec=True,
+ ),
+ patch.object(backend, "cleanup", autospec=True),
):
- cleanup_task_state_store(args)
+ clean_state_store(args)
backend.cleanup.assert_called_once_with()
@@ -44,22 +49,29 @@ class TestStateStoreCleanupCommand:
args = Namespace(dry_run=True, verbose=False)
backend = MetastoreBackend()
with (
- mock.patch("airflow.cli.commands.state_store_command.get_state_backend", return_value=backend),
- patch.object(backend, "_summary_dry_run", return_value={"expired": []}),
+ mock.patch(
+ "airflow.cli.commands.state_store_command.get_state_backend",
+ return_value=backend,
+ autospec=True,
+ ),
+ patch.object(backend, "_summary_dry_run", return_value={"expired": []}, autospec=True),
):
- cleanup_task_state_store(args)
+ clean_state_store(args)
captured = capsys.readouterr()
assert "Nothing to delete" in captured.out
def test_custom_backend_is_skipped(self, capsys):
+ # Custom (non-metastore) backends are intentionally skipped; cleanup() must not run.
args = Namespace(dry_run=False, verbose=False)
- custom_backend = MagicMock(spec=[])
+ custom_backend = MagicMock(spec=BaseStoreBackend)
with mock.patch(
- "airflow.cli.commands.state_store_command.get_state_backend", return_value=custom_backend
+ "airflow.cli.commands.state_store_command.get_state_backend",
+ return_value=custom_backend,
+ autospec=True,
):
- cleanup_task_state_store(args)
+ clean_state_store(args)
captured = capsys.readouterr()
- assert "Custom backend configured" in captured.out
- assert not hasattr(custom_backend, "cleanup") or not custom_backend.cleanup.called
+ assert "skipping" in captured.out.lower()
+ custom_backend.cleanup.assert_not_called()