refactor(coordination): centralize distributed coordination in a single service - #43316
Conversation
✅ Deploy Preview for superset-docs-preview ready!
To edit notification comments on pull requests, go to your Netlify project configuration. |
fb0f815 to
d3112e4
Compare
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## master #43316 +/- ##
========================================
Coverage 66.83% 66.83%
========================================
Files 2876 2880 +4
Lines 164061 164215 +154
Branches 37860 37900 +40
========================================
+ Hits 109657 109761 +104
- Misses 52230 52265 +35
- Partials 2174 2189 +15
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
d3112e4 to
8264ac8
Compare
80e7a76 to
db35479
Compare
nytai
left a comment
nytai
left a comment
There was a problem hiding this comment.
pretty straightforward refactor, approve with a minor comment
|
The |
Code Review Agent Run #99bcd5Actionable Suggestions - 0Additional Suggestions - 4
Filtered by Review RulesBito filtered these suggestions based on rules created automatically for your feedback. Manage rules.
Review Details
Bito Usage GuideCommands Type the following command in the pull request comment and save the comment.
Refer to the documentation for additional commands. Configuration This repository uses Documentation & Help AI Code Review powered by |
…ved cache_manager and renamed set_value flags
…AL_ASYNC_QUERIES is enabled The deprecated GLOBAL_ASYNC_QUERIES_CACHE_BACKEND ships a populated default, so CoordinationService.get_backend() treated it as a live backend everywhere — making is_backend_defined() return True even with GAQ off and no Redis, and pushing all DistributedLock/GTF callers at a nonexistent Redis. Gate the legacy fallback on the GLOBAL_ASYNC_QUERIES feature flag (the only real signal that GAQ is in use).
Move SignalListener + the CoordinationBackend alias to coordination/types.py and the close_pubsub helper to coordination/utils.py, mirroring the superset/distributed_lock layout (types.py + utils.py) and the existing coordination/exceptions.py. __init__ re-exports the public names, so `from superset.coordination import ...` is unchanged.
…nit__ thin Addresses review feedback: no implementation in __init__.py. CoordinationService now lives in coordination/base.py (matching the base.py convention used elsewhere, e.g. commands/distributed_lock/base.py); __init__ only holds the package docstring and re-exports (CoordinationService, SignalListener, CoordinationBackendUnavailableError), so `from superset.coordination import ...` is unchanged.
Drop the re-exports and __all__ from superset/coordination/__init__.py so the package init carries only the docstring, per review feedback that we don't use the __init__ re-export pattern here. All consumers now import directly from the submodule that owns the symbol: - CoordinationService -> superset.coordination.base - SignalListener -> superset.coordination.types - CoordinationBackendUnavailableError -> superset.coordination.exceptions Test patch targets and docstring cross-refs updated to the .base. paths.
…_signal fast path Addresses review feedback on apache#43316. get_backend() now resolves DISTRIBUTED_COORDINATION_CONFIG only. Previously it fell back to the deprecated GLOBAL_ASYNC_QUERIES_CACHE_BACKEND for *all* callers, so a deployment with GAQ enabled but no DISTRIBUTED_COORDINATION_CONFIG silently moved distributed locks (DB -> GAQ Redis) and GTF (DB polling -> GAQ Redis pub/sub) onto the GAQ backend — a rolling-upgrade split-brain for non-GAQ consumers (OAuth2/thumbnails/reports locks included). Locks and GTF are now byte-for-byte master behavior (DB when no coordinator is configured). Global Async Queries keep their own separate backend during the deprecation window: AsyncQueryManager resolves it (get_cache_backend, preferring the dedicated GLOBAL_ASYNC_QUERIES_CACHE_BACKEND and otherwise the coordinator) and passes it to the CoordinationService primitives via a new optional backend= param. The dual-backend arrangement is deprecated and collapses to DISTRIBUTED_COORDINATION_CONFIG in 8.0 (config.py + UPDATING.md). wait_for_signal now runs check() once before opening the pub/sub subscription and returns immediately if satisfied, so an already-terminal task never requires Redis to be reachable (and skips an unnecessary round-trip).
2bcd31a to
947ce35
Compare
![codeant-ai-for-open-source[bot]](https://avatars.githubusercontent.com/u/151810221?s=60&v=4){kind=small}
…end only as fallback Refines the GAQ backend resolution: AsyncQueryManager now uses the shared coordinator (DISTRIBUTED_COORDINATION_CONFIG) whenever it is configured, and only falls back to its dedicated (deprecated) GLOBAL_ASYNC_QUERIES_CACHE_BACKEND when the coordinator is unset — emitting the one-time deprecation warning only in that fallback case. A deployment that configures DISTRIBUTED_COORDINATION_CONFIG can thus retire the separate GAQ backend instead of maintaining two configs (also covers the Helm chart, which renders a GAQ backend block whenever cache is enabled). Non-breaking: the shipped GLOBAL_ASYNC_QUERIES_CACHE_BACKEND default is unchanged, so enabling GAQ without a coordinator still works. Docs (config.py, UPDATING.md, base.py) updated.
Code Review Agent Run #2adb26Actionable Suggestions - 0Additional Suggestions - 3
Filtered by Review RulesBito filtered these suggestions based on rules created automatically for your feedback. Manage rules.
Review Details
Bito Usage GuideCommands Type the following command in the pull request comment and save the comment.
Refer to the documentation for additional commands. Configuration This repository uses Documentation & Help AI Code Review powered by |
close_pubsub now attempts unsubscribe() and close() independently, so a failing unsubscribe no longer skips close() and leak the underlying Redis/pub-sub connection (the helper runs from every listener teardown and wait_for_signal finally block). Also corrects the CoordinationService docstring: the await/notify layer wakes on a published message when a backend is defined and polls only when none is defined; it does not poll as an outage fallback once a backend is configured.
The wait_for_signal fast-path fix evaluates check() once before opening the pub/sub subscription, so the mocked tests that assumed the first check happens after subscribe needed one extra predicate value: test_wait_for_signal_wakes_via_pubsub_and_cleans_up (add a None so a pub/sub nudge still occurs) and TaskManager's test_pubsub_success_subscribes_and_cleans_up (add a pending read so we still subscribe).
The write is skipped only when no coordination backend is configured; when one is configured a write failure propagates and fails job submission (it is not swallowed). The prior 'best-effort' wording overstated the guarantee.
…ocstring is_job_cancelled was the one GAQ method missing the self._gaq_backend is None guard, so with no dedicated backend it would fall through _require_backend() to the shared coordinator — the cross-backend resolution GAQ is meant to avoid. Restores the guard (matching the pre-refactor behavior of returning False). Also corrects the CoordinationService class docstring to note the primitives accept an explicit backend rather than always using the shared connection.
|
Went through this properly before approving, not just trusting the bot summaries. Good consolidation overall, and every open review thread turned out to already be fixed in code (moved One real blocker though: CI's actually red here, not flaky. One optional follow-up for later, not blocking: CodeAnt flagged that a mid-wait Redis outage in |
The GAQ backend must be resolved under the get_cache_backend patch, but init_app also re-registers the after_request handler and must run before login(): with login() first, its request tripped Flask's 'setup method after first request' guard before init_app re-registered the handler, failing all 6 TestAsyncEventApi cases. Resolve the mock under the patch during init_app (the backend is captured there), then log in and run.
Code Review Agent Run #5d3cb5Actionable Suggestions - 0Review Details
Bito Usage GuideCommands Type the following command in the pull request comment and save the comment.
Refer to the documentation for additional commands. Configuration This repository uses Documentation & Help AI Code Review powered by |
4 participants
SUMMARY
Introduces
superset.coordination.CoordinationService, a single entry point for the Valkey/Redis coordination primitives Superset relies on, and moves the repeated "publish a signal / await a signal with a timeout, else poll" business logic into it so consumers stay thin.Previously these were wired up ad hoc: the Global Task Framework (GTF) used
DISTRIBUTED_COORDINATION_CONFIGfor pub/sub + locking, Global Async Queries (GAQ) used a separateGLOBAL_ASYNC_QUERIES_CACHE_BACKENDfor its event streams, and the distributed lock and GTF's wait/abort loops each reached into the raw backend directly. This PR consolidates them behind one service to modularize and simplify the architecture, and to pave the way for further cleanup of GAQ.Public API — method by method
CoordinationServiceis a stateless, class-method service. It resolves its backend fromDISTRIBUTED_COORDINATION_CONFIGon each call — this is the single source of truth for the coordinator's consumers (distributed locks, GTF, and future stream/pub-sub users). It does not consult the deprecatedGLOBAL_ASYNC_QUERIES_CACHE_BACKEND; GAQ owns its own backend resolution (see below) and passes it to the primitives via an optionalbackend=argument.Backend / availability
get_backend() -> RedisCacheBackend | RedisSentinelCacheBackend | None— resolve the shared coordinator backend fromDISTRIBUTED_COORDINATION_CONFIG, orNoneif unconfigured. Escape hatch for the rare caller that needs the native connection (e.g. a long-lived pub/sub subscription loop).is_backend_defined() -> bool— whether the coordinator backend is configured. Best-effort callers branch on this before invoking the backend-only ops below.Pub/Sub
publish(channel, message, backend=None) -> int— fire-and-forget publish; returns the subscriber count. Subscribing is intentionally not wrapped (it needs the native long-lived connection), so subscribers useget_backend()— or, more commonly, the higher-level await/notify methods below.Key/Value (backend-agnostic flag names, mapped to Redis
SET/DELunder the hood)get_value(key, backend=None) -> Any— return the raw (bytes) value atkey, orNoneif absent.set_value(key, value, ttl=None, if_absent=False, if_present=False, backend=None) -> bool | None— storevalue;ttlsets an expiry (seconds),if_absent= set-only-if-missing (SET NX),if_present= set-only-if-exists (SET XX). ReturnsTrue, orNonewhen a condition prevented the write.delete_value(*keys, backend=None) -> int— delete one or more keys; returns the number deleted.Streams
stream_add(stream, data, event_id="*", max_len=None, backend=None) -> str— append an event to a stream; returns the generated event id.stream_range(stream, start="-", end="+", count=None, backend=None) -> list— read a range of events from a stream.Every raw primitive above accepts an optional
backend=so a caller with its own connection — GAQ, during the deprecation window — can run against it instead of the shared coordinator. Omitted (the default), they resolveget_backend()and raiseCoordinationBackendUnavailableErrorif no coordinator is configured.Await / notify (combine a pub/sub channel with a caller-supplied predicate; the message is only a wake-up nudge, the predicate is the source of truth)
wait_for_signal(channel, check, *, timeout=None, poll_interval=1.0) -> T— block untilcheck()returns a non-Nonevalue, and return it.check()is evaluated once before any subscription, so an already-satisfied wait returns straight from the predicate without touching the backend. When a backend is defined it then wakes promptly on a message published tochannel(re-checking the predicate each tick); when none is defined it pollscheckeverypoll_interval. RaisesTimeoutErroron timeout.listen_for_signal(channel, check, on_signal, *, poll_interval, name=None) -> SignalListener— run a background daemon that invokeson_signal()oncecheck()becomes true (woken by a published message when a backend is defined, else by polling). Returns a handle.Supporting types
SignalListener.stop()— signal the background listener to stop, close its subscription (so a thread parked inget_messagewakes immediately), and join.CoordinationBackendUnavailableError— raised by the backend-only ops (publish,*_value,stream_*) when no backend is available, instead of silently no-op'ing. Callers with their own fallback (e.g.DistributedLock's DB lock) gate onis_backend_defined()rather than catching it.Distributed locking is deliberately not on this service:
DistributedLock(superset/distributed_lock/) remains the user-facing lock interface and now draws on this service's backend when one is defined, falling back to a database-backed lock otherwise.Consumers refactored onto the service
DistributedLockacquire/release go throughset_value/delete_value(gated onis_backend_defined()) instead of reaching intocache_manager.distributed_coordination._cache; the redundantget_redis_client()helper is removed. Backend selection is unchanged from before this PR (coordinator only, DB fallback).TaskManager—wait_for_completion/listen_for_abortare now thin: they supply a channel + a task-status predicate (TERMINAL_STATES/ABORT_STATES) and delegate towait_for_signal/listen_for_signal;publish_abort/publish_completionusepublish. Removes ~250 lines of duplicated pub/sub-vs-poll branching and the bespokeAbortListener(nowSignalListener). Task edge cases (success/failure/abort/timeout) stay in GTF via the predicate over the authoritativeTaskrow. Backend selection is unchanged (coordinator only, DB polling fallback).AsyncQueryManagerresolves its own backend once atinit_app— the shared coordinator whenDISTRIBUTED_COORDINATION_CONFIGis set, otherwise its dedicated (deprecated)GLOBAL_ASYNC_QUERIES_CACHE_BACKEND— and passes it to the service's stream / KV primitives viabackend=, so GAQ's traffic stays scoped to the connection it resolved.Deprecation (non-breaking)
GLOBAL_ASYNC_QUERIES_CACHE_BACKENDis deprecated in favor ofDISTRIBUTED_COORDINATION_CONFIG. The two backends stay scoped during the deprecation window: distributed locks and GTF use the coordinator exclusively (falling back to the metadata database when it is unset, unchanged from before), and GAQ uses the coordinator whenever it is set — falling back to the dedicatedGLOBAL_ASYNC_QUERIES_CACHE_BACKEND(with a one-time deprecation warning) only when it is not. ConfiguringDISTRIBUTED_COORDINATION_CONFIGtherefore lets a deployment retire the separate GAQ backend rather than maintain two configs. All params are supported identically under the new key. The dual-backend arrangement is removed in Superset 8.0, when GAQ moves ontoDISTRIBUTED_COORDINATION_CONFIG.No functional change to async-query or task behavior; internal consolidation + config deprecation.
BEFORE/AFTER SCREENSHOTS OR ANIMATED GIF
N/A — internal refactor, no UI changes.
TESTING INSTRUCTIONS
pytest tests/unit_tests/coordination/test_service.py tests/unit_tests/distributed_lock/distributed_lock_tests.py tests/unit_tests/tasks/test_manager.py tests/unit_tests/async_events/async_query_manager_tests.pyDISTRIBUTED_COORDINATION_CONFIGset (Redis/Valkey): verify GTF abort/completion notifications, sync join-and-wait, distributed locking, and Global Async Queries all work end-to-end (GAQ rides the coordinator).GLOBAL_ASYNC_QUERIES_CACHE_BACKENDset (noDISTRIBUTED_COORDINATION_CONFIG): verify GAQ still works and a one-time deprecation warning is logged, while distributed locks / GTF use the metadata database.ADDITIONAL INFORMATION