refactor(coordination): centralize distributed coordination in a single service

UPDATING.md

@@ -61,6 +61,7 @@ the old counter to use the outcome-specific replacements.
 - [42393](https://github.com/apache/superset/pull/42393): Exported dataset YAML now carries a `uuid` for each metric and column so that custom folder assignments (which reference metrics/columns by UUID) survive an import into another workspace. This affects any export bundle that contains datasets, not just a dataset export: chart, dashboard, database and full-asset exports all embed the same dataset YAML, so a dashboard exported from this release also fails to import into an older one even though no dataset was exported directly. As with `folders` and `currency_code_column`, the affected `datasets/` files fail schema validation (`Unknown field: uuid`) when imported into Superset releases that predate this change; regenerate or hand-edit exports for older targets in mixed-version fleets.
 - [42300](https://github.com/apache/superset/pull/42300): Timeseries charts (line/area/bar) with a Y-axis bound in effect — either an explicit `yAxisBounds` or one derived from `truncateYAxis` — now clamp out-of-range data points to that bound instead of letting ECharts drop the point (and the line segments around it) entirely. Any existing chart with a configured Y-axis bound and data outside it will look different after upgrading: a gap becomes a point pinned to the boundary. The clamp also rewrites the value ECharts reads for that point's tooltip and data label, so the displayed value is the bound rather than the true observation.
 - [42087](https://github.com/apache/superset/pull/42087): Stored calculated-column and metric expressions are validated when a query is built, under the same sub-query policy already applied to adhoc expressions. Previously only the dataset update path checked them on save, so expressions written by v1 import, by dataset duplication, or before that check existed were never validated. Since `ALLOW_ADHOC_SUBQUERY` defaults to `False` (see [19242](https://github.com/apache/superset/pull/19242)), a dataset whose stored expression contains a sub-query works before upgrading and afterwards fails at chart render with `Custom SQL fields cannot contain sub-queries.` There is no migration step, and the error does not name the offending dataset column, so audit stored expressions before upgrading: either rewrite them without the sub-query, or set `ALLOW_ADHOC_SUBQUERY = True` to keep the previous behaviour for both stored and adhoc expressions.
+- The `GLOBAL_ASYNC_QUERIES_CACHE_BACKEND` config key is **deprecated** in favor of the shared coordination backend `DISTRIBUTED_COORDINATION_CONFIG`, which powers a single Redis connection for distributed locks, pub/sub, and the async-events streams (the GAQ firehose). All parameters previously accepted by `GLOBAL_ASYNC_QUERIES_CACHE_BACKEND` are supported identically under `DISTRIBUTED_COORDINATION_CONFIG` (they use the same `RedisCache`/`RedisSentinelCache` backend). During the deprecation window the two backends stay scoped: distributed locks and the Global Task Framework use `DISTRIBUTED_COORDINATION_CONFIG` exclusively (falling back to the metadata database when it is unset, unchanged from before), and Global Async Queries use `DISTRIBUTED_COORDINATION_CONFIG` whenever it is set — falling back to the dedicated `GLOBAL_ASYNC_QUERIES_CACHE_BACKEND` (with a one-time deprecation warning) only when it is not. Configuring `DISTRIBUTED_COORDINATION_CONFIG` therefore lets a deployment retire the separate `GLOBAL_ASYNC_QUERIES_CACHE_BACKEND` rather than maintain two configs. This dual-backend arrangement is removed in Superset 8.0, when GAQ moves onto `DISTRIBUTED_COORDINATION_CONFIG`; migrate now by configuring it.
 
 ### Selenium support removed — Playwright is now required for screenshots
 

docs/admin_docs/configuration/cache.mdx

@@ -345,9 +345,10 @@ DISTRIBUTED_COORDINATION_CONFIG = {
 }
 ```
 
-By default, connections opened for `DISTRIBUTED_COORDINATION_CONFIG` (as well as
-`GLOBAL_ASYNC_QUERIES_CACHE_BACKEND`, which uses the same `RedisCache`/`RedisSentinelCache`
-backend) have no socket timeout. This can be overridden with `CACHE_REDIS_SOCKET_TIMEOUT` and
+By default, connections opened for `DISTRIBUTED_COORDINATION_CONFIG` (as well as the
+deprecated `GLOBAL_ASYNC_QUERIES_CACHE_BACKEND`, which uses the same
+`RedisCache`/`RedisSentinelCache` backend) have no socket timeout. This can be
+overridden with `CACHE_REDIS_SOCKET_TIMEOUT` and
 `CACHE_REDIS_SOCKET_CONNECT_TIMEOUT`, both in seconds:
 
 ```python

superset/async_events/async_query_manager.py

@@ -25,12 +25,12 @@
 
 import jwt
 from flask import Flask, Request, request, Response, session
-from flask_caching.backends.base import BaseCache
 
 from superset.async_events.cache_backend import (
     RedisCacheBackend,
     RedisSentinelCacheBackend,
 )
+from superset.coordination.base import CoordinationService
 from superset.utils import json
 from superset.utils.core import get_user_id
 
@@ -87,6 +87,12 @@ def increment_id(entry_id: str) -> str:
 def get_cache_backend(
     config: dict[str, Any],
 ) -> RedisCacheBackend | RedisSentinelCacheBackend:
+    """Build a coordination backend from the deprecated GAQ cache config.
+
+    DEPRECATED: retained only so Global Async Queries can run on its own dedicated
+    ``GLOBAL_ASYNC_QUERIES_CACHE_BACKEND`` during the deprecation window. Removed in
+    Superset 8.0, when GAQ moves onto ``DISTRIBUTED_COORDINATION_CONFIG``.
+    """
     cache_config = config.get("GLOBAL_ASYNC_QUERIES_CACHE_BACKEND", {})
     cache_type = cache_config.get("CACHE_TYPE")
 
@@ -96,7 +102,6 @@ def get_cache_backend(
     if cache_type == "RedisSentinelCache":
         return RedisSentinelCacheBackend.from_config(cache_config)
 
-    # TODO: Expand cache backend options.
     raise UnsupportedCacheBackendError("Unsupported cache backend configuration")
 
 
@@ -110,13 +115,18 @@ class AsyncQueryManager:
     # Redis key prefix (within the GAQ stream namespace) for the per-job record
     # that authorizes cancellation and flags a job as cancelled for the worker.
     _JOB_REGISTRY_PREFIX = "job-cancel:"
+    # Emit the dedicated-backend deprecation notice at most once per process.
+    _legacy_backend_warning_emitted: bool = False
 
     def __init__(self) -> None:
         super().__init__()
-        self._cache: Optional[BaseCache] = None
         self._stream_prefix: str = ""
         self._stream_limit: Optional[int]
         self._stream_limit_firehose: Optional[int]
+        # Global Async Queries owns its coordination backend separately from the
+        # shared coordinator (see init_app); resolved there and passed explicitly
+        # to CoordinationService's primitives.
+        self._gaq_backend: RedisCacheBackend | RedisSentinelCacheBackend | None = None
         self._jwt_cookie_name: str = ""
         self._jwt_cookie_secure: bool = False
         self._jwt_cookie_domain: Optional[str]
@@ -137,8 +147,42 @@ def init_app(self, app: Flask) -> None:
                 """
             )
 
-        self._cache = get_cache_backend(app.config)
-        logger.debug("Using GAQ Cache backend as %s", type(self._cache).__name__)
+        if not (
+            app.config.get("DISTRIBUTED_COORDINATION_CONFIG")
+            or app.config.get("GLOBAL_ASYNC_QUERIES_CACHE_BACKEND", {}).get(
+                "CACHE_TYPE"
+            )
+        ):
+            raise UnsupportedCacheBackendError(
+                "Global async queries require a coordination backend; configure "
+                "DISTRIBUTED_COORDINATION_CONFIG (GLOBAL_ASYNC_QUERIES_CACHE_BACKEND "
+                "is deprecated)."
+            )
+
+        # Global Async Queries share the coordinator's backend whenever
+        # DISTRIBUTED_COORDINATION_CONFIG is configured, so an 8.0-leaning deployment
+        # keeps a single coordination config rather than maintaining a separate one.
+        # Only when no coordinator is configured does GAQ fall back to its dedicated
+        # (deprecated) GLOBAL_ASYNC_QUERIES_CACHE_BACKEND, emitting a one-time
+        # deprecation warning. That dedicated backend is removed in Superset 8.0, when
+        # DISTRIBUTED_COORDINATION_CONFIG becomes the only option. Either way GAQ passes
+        # its resolved backend explicitly to CoordinationService's primitives, keeping
+        # its stream/pub-sub traffic scoped to the connection it resolved here.
+        if (coordinator := CoordinationService.get_backend()) is not None:
+            self._gaq_backend = coordinator
+        else:
+            if not AsyncQueryManager._legacy_backend_warning_emitted:
+                logger.warning(
+                    "Global Async Queries is running on the deprecated "
+                    "GLOBAL_ASYNC_QUERIES_CACHE_BACKEND because "
+                    "DISTRIBUTED_COORDINATION_CONFIG is not configured. Configure "
+                    "DISTRIBUTED_COORDINATION_CONFIG to consolidate coordination "
+                    "(distributed locks, task framework, pub/sub, and the GAQ event "
+                    "streams) onto a single connection; the dedicated backend is "
+                    "removed in Superset 8.0."
+                )
+                AsyncQueryManager._legacy_backend_warning_emitted = True
+            self._gaq_backend = get_cache_backend(app.config)
 
         if len(app.config["GLOBAL_ASYNC_QUERIES_JWT_SECRET"]) < 32:
             raise AsyncQueryTokenException(
@@ -298,12 +342,17 @@ def _register_cancellable_job(
         owner without trusting the client-supplied id. Expires with the JWT so
         it never outlives the job it guards.
         """
-        if not self._cache:
+        # The cancel registry is an optimization: skip it when GAQ has no
+        # coordination backend configured. When a backend is configured, a write
+        # failure is not swallowed here — it surfaces to the caller and fails job
+        # submission, matching the behavior of the surrounding stream writes.
+        if self._gaq_backend is None:
             return
-        self._cache.set(
+        CoordinationService.set_value(
             self._job_registry_key(job_id),
             json.dumps({"channel_id": channel_id, "user_id": user_id}),
-            ex=self._jwt_expiration_seconds or None,
+            ttl=self._jwt_expiration_seconds or None,
+            backend=self._gaq_backend,
         )
 
     def submit_chart_data_job(
@@ -337,33 +386,32 @@ def submit_chart_data_job(
     def read_events(
         self, channel: str, last_id: Optional[str]
     ) -> list[Optional[dict[str, Any]]]:
-        if not self._cache:
+        if self._gaq_backend is None:
             raise CacheBackendNotInitialized("Cache backend not initialized")
 
         stream_name = f"{self._stream_prefix}{channel}"
         start_id = increment_id(last_id) if last_id else "-"
-        results = self._cache.xrange(stream_name, start_id, "+", self.MAX_EVENT_COUNT)
-        # Decode bytes to strings, decode_responses is not supported at RedisCache and RedisSentinelCache  # noqa: E501
-        if isinstance(self._cache, (RedisSentinelCacheBackend, RedisCacheBackend)):
-            decoded_results = [
-                (
-                    event_id.decode("utf-8"),
-                    {
-                        key.decode("utf-8"): value.decode("utf-8")
-                        for key, value in event_data.items()
-                    },
-                )
-                for event_id, event_data in results
-            ]
-            return (
-                [] if not decoded_results else list(map(parse_event, decoded_results))
+        results = CoordinationService.stream_range(
+            stream_name, start_id, "+", self.MAX_EVENT_COUNT, backend=self._gaq_backend
+        )
+        # Decode bytes to strings: the coordination Redis backends do not enable
+        # decode_responses, so stream_range returns raw bytes.
+        decoded_results = [
+            (
+                event_id.decode("utf-8"),
+                {
+                    key.decode("utf-8"): value.decode("utf-8")
+                    for key, value in event_data.items()
+                },
             )
-        return [] if not results else list(map(parse_event, results))
+            for event_id, event_data in results
+        ]
+        return [] if not decoded_results else list(map(parse_event, decoded_results))
 
     def update_job(
         self, job_metadata: dict[str, Any], status: str, **kwargs: Any
     ) -> None:
-        if not self._cache:
+        if self._gaq_backend is None:
             raise CacheBackendNotInitialized("Cache backend not initialized")
 
         if "channel_id" not in job_metadata:
@@ -388,10 +436,24 @@ def update_job(
         # SIGUSR1 it is about to receive as a cancellation.
         if status in (self.STATUS_DONE, self.STATUS_ERROR):
             if job_id := job_metadata.get("job_id"):
-                self._cache.delete(self._job_registry_key(job_id))
+                CoordinationService.delete_value(
+                    self._job_registry_key(job_id), backend=self._gaq_backend
+                )
 
-        self._cache.xadd(scoped_stream_name, event_data, "*", self._stream_limit)
-        self._cache.xadd(full_stream_name, event_data, "*", self._stream_limit_firehose)
+        CoordinationService.stream_add(
+            scoped_stream_name,
+            event_data,
+            "*",
+            self._stream_limit,
+            backend=self._gaq_backend,
+        )
+        CoordinationService.stream_add(
+            full_stream_name,
+            event_data,
+            "*",
+            self._stream_limit_firehose,
+            backend=self._gaq_backend,
+        )
 
     def is_job_cancelled(self, job_id: str) -> bool:
         """
@@ -401,10 +463,12 @@ def is_job_cancelled(self, job_id: str) -> bool:
         swallowed and treated as "not cancelled" — a Redis blip must never mask
         the original error (e.g. a genuine timeout) with a connection error.
         """
-        if not self._cache:
+        if self._gaq_backend is None:
             return False
         try:
-            raw = self._cache.get(self._job_registry_key(job_id))
+            raw = CoordinationService.get_value(
+                self._job_registry_key(job_id), backend=self._gaq_backend
+            )
             if raw is None:
                 return False
             return bool(json.loads(raw).get("cancelled"))
@@ -428,11 +492,11 @@ def cancel_job(self, job_id: str, channel_id: str, user_id: Optional[int]) -> No
         :raises AsyncQueryJobException: the job is unknown or already terminal
         :raises AsyncQueryTokenException: the caller does not own the job
         """
-        if not self._cache:
+        if self._gaq_backend is None:
             raise CacheBackendNotInitialized("Cache backend not initialized")
 
         key = self._job_registry_key(job_id)
-        raw = self._cache.get(key)
+        raw = CoordinationService.get_value(key, backend=self._gaq_backend)
         if raw is None:
             raise AsyncQueryJobException("Job not found or already completed")
 
@@ -445,11 +509,12 @@ def cancel_job(self, job_id: str, channel_id: str, user_id: Optional[int]) -> No
         # key still exists (``xx``): if the job finished and cleared its record
         # between the read above and here, don't recreate a stale record or
         # revoke a task that is already gone — report it as not found instead.
-        flagged = self._cache.set(
+        flagged = CoordinationService.set_value(
             key,
             json.dumps({**record, "cancelled": True}),
-            ex=self._jwt_expiration_seconds or None,
-            xx=True,
+            ttl=self._jwt_expiration_seconds or None,
+            if_present=True,
+            backend=self._gaq_backend,
         )
         if not flagged:
             raise AsyncQueryJobException("Job not found or already completed")

superset/commands/distributed_lock/acquire.py

@@ -27,8 +27,8 @@
 from superset.commands.distributed_lock.base import (
     BaseDistributedLockCommand,
     get_default_lock_ttl,
-    get_redis_client,
 )
+from superset.coordination.base import CoordinationService
 from superset.daos.key_value import KeyValueDAO
 from superset.exceptions import (
     AcquireDistributedLockFailedException,
@@ -80,21 +80,21 @@ def __init__(
         self.ttl_seconds = ttl_seconds or get_default_lock_ttl()
 
     def run(self) -> None:
-        if (redis_client := get_redis_client()) is not None:
-            self._acquire_redis(redis_client)
+        if CoordinationService.is_backend_defined():
+            self._acquire_redis()
         else:
             self._acquire_kv()
 
-    def _acquire_redis(self, redis_client: Any) -> None:
-        """Acquire lock using Redis SET NX EX (atomic)."""
+    def _acquire_redis(self) -> None:
+        """Acquire lock using the coordination backend's SET NX EX (atomic)."""
         try:
             # SET NX EX: Set if not exists, with expiration
             # Returns True if lock acquired, None if already exists
-            acquired = redis_client.set(
+            acquired = CoordinationService.set_value(
                 self.redis_lock_key,
                 "1",
-                nx=True,
-                ex=self.ttl_seconds,
+                ttl=self.ttl_seconds,
+                if_absent=True,
             )
 
             if not acquired:

superset/commands/distributed_lock/base.py

@@ -19,18 +19,14 @@
 
 import logging
 import uuid
-from typing import Any, TYPE_CHECKING
+from typing import Any
 
 from flask import current_app
 
 from superset.commands.base import BaseCommand
 from superset.distributed_lock.utils import get_key
-from superset.extensions import cache_manager
 from superset.key_value.types import JsonKeyValueCodec, KeyValueResource
 
-if TYPE_CHECKING:
-    import redis
-
 logger = logging.getLogger(__name__)
 
 
@@ -39,17 +35,6 @@ def get_default_lock_ttl() -> int:
     return int(current_app.config.get("DISTRIBUTED_LOCK_DEFAULT_TTL", 30))
 
 
-def get_redis_client() -> "redis.Redis[Any] | None":
-    """
-    Get Redis client from distributed coordination if available.
-
-    Returns None if DISTRIBUTED_COORDINATION_CONFIG is not configured,
-    allowing fallback to database-backed locking.
-    """
-    backend = cache_manager.distributed_coordination
-    return backend._cache if backend else None
-
-
 class BaseDistributedLockCommand(BaseCommand):
     """Base command for distributed lock operations."""
 

superset/commands/distributed_lock/release.py

@@ -19,15 +19,12 @@
 
 import logging
 from functools import partial
-from typing import Any
 
 import redis
 from sqlalchemy.exc import SQLAlchemyError
 
-from superset.commands.distributed_lock.base import (
-    BaseDistributedLockCommand,
-    get_redis_client,
-)
+from superset.commands.distributed_lock.base import BaseDistributedLockCommand
+from superset.coordination.base import CoordinationService
 from superset.daos.key_value import KeyValueDAO
 from superset.exceptions import ReleaseDistributedLockFailedException
 from superset.key_value.exceptions import KeyValueDeleteFailedError
@@ -45,15 +42,15 @@ class ReleaseDistributedLock(BaseDistributedLockCommand):
     """
 
     def run(self) -> None:
-        if (redis_client := get_redis_client()) is not None:
-            self._release_redis(redis_client)
+        if CoordinationService.is_backend_defined():
+            self._release_redis()
         else:
             self._release_kv()
 
-    def _release_redis(self, redis_client: Any) -> None:
-        """Release lock using Redis DELETE."""
+    def _release_redis(self) -> None:
+        """Release lock using the coordination backend's DELETE."""
         try:
-            redis_client.delete(self.redis_lock_key)
+            CoordinationService.delete_value(self.redis_lock_key)
             logger.debug("Released Redis lock: %s", self.redis_lock_key)
         except redis.RedisError as ex:
             # Log warning but don't raise - TTL will handle cleanup