This is an automated email from the ASF dual-hosted git repository.
Alanxtl pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/dubbo-go.git
The following commit(s) were added to refs/heads/develop by this push:
new 2046d8dec Feat: Add ZK config cache (#3612)
2046d8dec is described below
commit 2046d8decbf889e7676be963cdad5e74d1bf66fd
Author: wm_03 <[email protected]>
AuthorDate: Tue Sep 1 17:13:17 2026 +0800
Feat: Add ZK config cache (#3612)
* feat: add Zookeeper config cache
* test: cover Zookeeper cache and watch flows
* refactor(zookeeper): bound config cache path locks
* fix(zookeeper): bound config cache entries
* refactor(zookeeper): track watcher ownership
* fix(zookeeper): bound automatic watches
* fix(zookeeper): complete config watch lifecycle
* fix(zookeeper): align listener paths and empty config events
* fix(zookeeper): restore and clean up watches on reconnect
* fix(zookeeper): safely remove expired cache entries
* fix(zookeeper): restore business watches with cache disabled
* fix(zookeeper): release configuration listener wait group
* ci: configure ZooKeeper dependency for tests
* fix(zookeeper): adapt config watches to upstream API
* test(zookeeper): cover session-aware watch registration
* test(zookeeper): cover retired watch lifecycle
* fix(zookeeper): restore watches across sessions
* fix(zookeeper): synchronize config watch lifecycle
* fix(zookeeper): synchronize watch ownership and retries
* fix(zookeeper): reject stale watch results
* fix(zookeeper): synchronize listener watch retries
* fix: Add the TTL upper and lower bounds
* annotations: Add relevant annotations for the core concepts
---
.github/workflows/github-actions.yml | 26 +
common/constant/key.go | 1 +
config_center/zookeeper/config_cache.go | 703 +++++++++++++++++++++++++++
config_center/zookeeper/config_cache_test.go | 572 ++++++++++++++++++++++
config_center/zookeeper/impl.go | 150 +++++-
config_center/zookeeper/impl_test.go | 338 +++++++++++++
config_center/zookeeper/listener.go | 345 ++++++++++++-
config_center/zookeeper/listener_test.go | 452 ++++++++++++++++-
remoting/zookeeper/listener.go | 186 +++++--
remoting/zookeeper/listener_test.go | 207 ++++++++
10 files changed, 2892 insertions(+), 88 deletions(-)
diff --git a/.github/workflows/github-actions.yml
b/.github/workflows/github-actions.yml
index 8fec237c7..13bdce900 100644
--- a/.github/workflows/github-actions.yml
+++ b/.github/workflows/github-actions.yml
@@ -48,6 +48,32 @@ jobs:
- name: Setup Go
uses: ./.github/actions/setup-go
+ - name: Setup Java for ZooKeeper tests
+ uses: actions/setup-java@v4
+ with:
+ distribution: temurin
+ java-version: '8'
+
+ - name: Prepare ZooKeeper test dependency
+ run: |
+ set -euo pipefail
+
+ go mod download github.com/dubbogo/gost
+
+ gost_dir="$(go list -m -f '{{.Dir}}' github.com/dubbogo/gost)"
+ jar_path="$(find "$gost_dir/database/kv/zk" \
+ -type f \
+ -path '*/contrib/fatjar/zookeeper-*-fatjar.jar' \
+ -print -quit)"
+
+ test -n "$jar_path"
+ java -version
+
+ zk_path="${jar_path%/contrib/fatjar/*}"
+ test -f "$jar_path"
+
+ echo "ZOOKEEPER_PATH=$zk_path" >> "$GITHUB_ENV"
+
- name: Run unit tests
run: make test
diff --git a/common/constant/key.go b/common/constant/key.go
index bcd35c26a..8b24dbde3 100644
--- a/common/constant/key.go
+++ b/common/constant/key.go
@@ -261,6 +261,7 @@ const (
ConfigSecretKey = "config-center.secret"
ConfigBackupConfigKey = "config-center.isBackupConfig"
ConfigBackupConfigPathKey = "config-center.backupConfigPath"
+ ConfigCacheTTLKey = "config-center.cache-ttl"
ConfigRootPathParamKey = "dubbo.config-center.root-path"
)
diff --git a/config_center/zookeeper/config_cache.go
b/config_center/zookeeper/config_cache.go
new file mode 100644
index 000000000..a3d525dea
--- /dev/null
+++ b/config_center/zookeeper/config_cache.go
@@ -0,0 +1,703 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package zookeeper
+
+import (
+ "errors"
+ "sync"
+ "time"
+)
+
+import (
+ "github.com/go-zookeeper/zk"
+
+ "github.com/hashicorp/golang-lru"
+)
+
+const (
+ pathLockShardCount = 128
+ maxCacheEntries = 1024
+ maxAutoWatches = 1024
+ maxCacheLoadRetries = 3
+)
+
+var errWatchRegistrationStale = errors.New("zookeeper watch registration
became stale")
+var errBusinessWatchCanceled = errors.New("zookeeper business watch
registration canceled")
+
+// watchOperation represents one in-flight watch registration. Concurrent
+// callers for the same path wait on done instead of registering another watch,
+// while token prevents a late result from completing a newer operation.
+type watchOperation struct {
+ token uint64
+ done chan struct{}
+ err error
+ completed bool
+}
+
+// configCacheEntry keeps node existence separate from content so a missing
+// node is distinguishable from an existing node whose content is empty.
+type configCacheEntry struct {
+ content string
+ exists bool
+ expiresAt time.Time
+}
+
+// watchRegistration captures the event channel returned by ZooKeeper and the
+// sessions around both watch registration and its optional follow-up read. A
+// result is usable only when these operations belong to one stable session.
+type watchRegistration struct {
+ events <-chan zk.Event
+ beforeSessionID int64
+ afterSessionID int64
+ readBeforeSessionID int64
+ readAfterSessionID int64
+}
+
+// resolve reports the session associated with the registration. If
registration
+// crosses a session boundary, it also rejects a watch whose channel has
already
+// received an event.
+func (r watchRegistration) resolve() (int64, bool) {
+ if r.events == nil || r.afterSessionID == 0 {
+ return 0, false
+ }
+ if r.beforeSessionID != r.afterSessionID {
+ select {
+ case <-r.events:
+ return 0, false
+ default:
+ }
+ }
+ return r.resultSessionID(), true
+}
+
+// sessionStable reports whether registration and any follow-up read stayed in
+// the same ZooKeeper session. Zero pairs are allowed for operations that did
+// not access a connection, but all observed non-zero IDs must agree.
+func (r watchRegistration) sessionStable() bool {
+ if !stableSessionPair(r.beforeSessionID, r.afterSessionID) ||
+ !stableSessionPair(r.readBeforeSessionID, r.readAfterSessionID)
{
+ return false
+ }
+ var sessionID int64
+ for _, id := range []int64{
+ r.beforeSessionID,
+ r.afterSessionID,
+ r.readBeforeSessionID,
+ r.readAfterSessionID,
+ } {
+ if id == 0 {
+ continue
+ }
+ if sessionID == 0 {
+ sessionID = id
+ continue
+ }
+ if sessionID != id {
+ return false
+ }
+ }
+ return true
+}
+
+func (r watchRegistration) resultSessionID() int64 {
+ if r.readAfterSessionID != 0 {
+ return r.readAfterSessionID
+ }
+ return r.afterSessionID
+}
+
+func stableSessionPair(beforeSessionID, afterSessionID int64) bool {
+ return (beforeSessionID == 0 && afterSessionID == 0) ||
+ (beforeSessionID != 0 && beforeSessionID == afterSessionID)
+}
+
+// configWatchState tracks the lifecycle and ownership of one concrete-path
+// watch. Auto watches accelerate cache updates and consume the bounded auto
+// quota; business watches are retained while configuration listeners exist.
+type configWatchState struct {
+ registered bool // ZooKeeper accepted the watch registration.
+ pending bool // A registration is in flight and reserves
ownership.
+ auto bool // The cache, rather than a business
listener, owns the watch.
+ retired bool // Business ownership ended, but no auto
quota was available.
+ sessionID int64 // ZooKeeper session associated with the
registered or pending watch.
+ pendingOp *watchOperation // Shared completion state for the in-flight
registration.
+}
+
+func (s configWatchState) tracked() bool {
+ return s.registered || s.pending
+}
+
+func (s configWatchState) holdsAutoSlot() bool {
+ return s.auto && s.tracked()
+}
+
+func (s configWatchState) holdsAutoWatch() bool {
+ return s.auto && s.registered
+}
+
+func (s configWatchState) holdsAutoReservation() bool {
+ return s.auto && s.pending
+}
+
+func (s configWatchState) pendingOpToken() uint64 {
+ if s.pendingOp == nil {
+ return 0
+ }
+ return s.pendingOp.token
+}
+
+// configCache combines a bounded read-through cache with the concrete-path
+// watch state needed to keep cached values current between TTL refreshes.
+type configCache struct {
+ ttl time.Duration
+
+ stateLock sync.RWMutex
+ entries *lru.Cache // Bounded LRU containing both
existing and missing nodes.
+ watches map[string]configWatchState
+ autoWatchCount int
+ autoWatchReservations int
+ generation uint64 // Incremented on reset to reject results
from earlier cache epochs.
+ sessionID int64 // Current ZooKeeper session observed by
the latest reset.
+ nextWatchToken uint64
+
+ // Fixed shards serialize cache and listener transitions for the same
path
+ // without retaining one lock for every path ever requested.
+ pathLocks [pathLockShardCount]sync.Mutex
+}
+
+func newConfigCache(ttl time.Duration) configCache {
+ entries, err := lru.New(maxCacheEntries)
+ if err != nil {
+ panic(err)
+ }
+ return configCache{
+ ttl: ttl,
+ entries: entries,
+ watches: make(map[string]configWatchState),
+ }
+}
+
+func (c *configCache) enabled() bool {
+ return c.ttl > 0
+}
+
+func (c *configCache) load(
+ path string,
+ loader func(bool) (configCacheEntry, watchRegistration, error),
+) (configCacheEntry, error) {
+ if !c.enabled() {
+ generation, _ := c.snapshot(path)
+ entry, registration, err := loader(false)
+ if err == nil && !c.loadResultCurrent(generation, registration)
{
+ return configCacheEntry{}, errWatchRegistrationStale
+ }
+ return entry, err
+ }
+ if entry, ok := c.getFresh(path); ok {
+ return entry, nil
+ }
+
+ pathLock := c.pathLock(path)
+ pathLock.Lock()
+ defer pathLock.Unlock()
+
+ watchRetries := 0
+ for attempt := 0; attempt <= maxCacheLoadRetries; attempt++ {
+ if entry, ok := c.getFresh(path); ok {
+ return entry, nil
+ }
+
+ generation, registerWatch, token := c.prepareLoad(path)
+ entry, registration, err := loader(registerWatch)
+ if registerWatch {
+ stored := c.finishWatchRegistrationLocked(path,
generation, token, registration)
+ if !stored || !c.loadResultCurrent(generation,
registration) {
+ if watchRetries == 0 && attempt <
maxCacheLoadRetries {
+ watchRetries++
+ continue
+ }
+ fallbackEntry, fallbackRegistration,
fallbackErr := loader(false)
+ if fallbackErr != nil {
+ return configCacheEntry{}, fallbackErr
+ }
+ if !c.loadResultCurrent(generation,
fallbackRegistration) {
+ if attempt == maxCacheLoadRetries {
+ return configCacheEntry{},
errWatchRegistrationStale
+ }
+ continue
+ }
+ if c.storeLoadEntry(path, generation,
fallbackEntry) {
+ return fallbackEntry, nil
+ }
+ if attempt == maxCacheLoadRetries {
+ return configCacheEntry{},
errWatchRegistrationStale
+ }
+ continue
+ }
+ } else if !c.loadResultCurrent(generation, registration) {
+ if attempt == maxCacheLoadRetries {
+ return configCacheEntry{},
errWatchRegistrationStale
+ }
+ continue
+ }
+
+ if err != nil {
+ if attempt == maxCacheLoadRetries &&
!c.isCurrentGeneration(generation) {
+ return configCacheEntry{},
errWatchRegistrationStale
+ }
+ if !c.isCurrentGeneration(generation) {
+ continue
+ }
+ return configCacheEntry{}, err
+ }
+
+ if c.storeLoadEntry(path, generation, entry) {
+ return entry, nil
+ }
+ if attempt == maxCacheLoadRetries {
+ return configCacheEntry{}, errWatchRegistrationStale
+ }
+ }
+ return configCacheEntry{}, errWatchRegistrationStale
+}
+
+func (c *configCache) prepareLoad(path string) (uint64, bool, uint64) {
+ c.stateLock.Lock()
+ defer c.stateLock.Unlock()
+
+ generation := c.generation
+ if c.watches[path].tracked() {
+ return generation, false, 0
+ }
+
+ op := c.newWatchOperationLocked()
+ pendingState := configWatchState{auto: true, pending: true, sessionID:
c.sessionID, pendingOp: op}
+ if !c.setWatchStateLocked(path, pendingState) {
+ return generation, false, 0
+ }
+ return generation, true, op.token
+}
+
+func (c *configCache) store(path string, entry configCacheEntry) {
+ if !c.enabled() {
+ return
+ }
+ pathLock := c.pathLock(path)
+ pathLock.Lock()
+ defer pathLock.Unlock()
+
+ c.stateLock.Lock()
+ defer c.stateLock.Unlock()
+ c.storeEntryLocked(path, entry)
+}
+
+func (c *configCache) storeAtGenerationLocked(path string, generation uint64,
entry configCacheEntry) {
+ if !c.enabled() {
+ return
+ }
+ c.stateLock.Lock()
+ defer c.stateLock.Unlock()
+ if c.generation == generation {
+ c.storeEntryLocked(path, entry)
+ }
+}
+
+func (c *configCache) storeLocked(path string, entry configCacheEntry) {
+ if !c.enabled() {
+ return
+ }
+ c.stateLock.Lock()
+ defer c.stateLock.Unlock()
+ c.storeEntryLocked(path, entry)
+}
+
+func (c *configCache) getFresh(path string) (configCacheEntry, bool) {
+ c.stateLock.RLock()
+ value, ok := c.entries.Get(path)
+ if !ok {
+ c.stateLock.RUnlock()
+ return configCacheEntry{}, false
+ }
+ entry := value.(configCacheEntry)
+ if entry.expiresAt.After(time.Now()) {
+ c.stateLock.RUnlock()
+ return entry, true
+ }
+ c.stateLock.RUnlock()
+
+ c.stateLock.Lock()
+ defer c.stateLock.Unlock()
+ value, ok = c.entries.Get(path)
+ if !ok {
+ return configCacheEntry{}, false
+ }
+ entry = value.(configCacheEntry)
+ if entry.expiresAt.After(time.Now()) {
+ return entry, true
+ }
+ c.entries.Remove(path)
+ return configCacheEntry{}, false
+}
+
+func (c *configCache) storeEntryLocked(path string, entry configCacheEntry) {
+ entry.expiresAt = time.Now().Add(c.ttl)
+ c.entries.Add(path, entry)
+}
+
+func (c *configCache) snapshot(path string) (uint64, configWatchState) {
+ c.stateLock.RLock()
+ defer c.stateLock.RUnlock()
+ return c.generation, c.watches[path]
+}
+
+func (c *configCache) isCurrentGeneration(generation uint64) bool {
+ c.stateLock.RLock()
+ defer c.stateLock.RUnlock()
+ return c.generation == generation
+}
+
+// loadResultCurrent is the final fence before a ZooKeeper read can affect the
+// cache: both the cache generation and every observed session must be current.
+func (c *configCache) loadResultCurrent(generation uint64, registration
watchRegistration) bool {
+ if !registration.sessionStable() {
+ return false
+ }
+ c.stateLock.RLock()
+ defer c.stateLock.RUnlock()
+ if c.generation != generation {
+ return false
+ }
+ resultSessionID := registration.resultSessionID()
+ return c.sessionID == 0 ||
+ (resultSessionID != 0 && resultSessionID == c.sessionID)
+}
+
+func (c *configCache) storeLoadEntry(path string, generation uint64, entry
configCacheEntry) bool {
+ c.stateLock.Lock()
+ defer c.stateLock.Unlock()
+ if c.generation != generation {
+ return false
+ }
+ c.storeEntryLocked(path, entry)
+ return true
+}
+
+func (c *configCache) newWatchOperationLocked() *watchOperation {
+ c.nextWatchToken++
+ return &watchOperation{
+ token: c.nextWatchToken,
+ done: make(chan struct{}),
+ }
+}
+
+func (c *configCache) completeWatchOperationLocked(op *watchOperation, err
error) {
+ if op == nil || op.completed {
+ return
+ }
+ op.err = err
+ op.completed = true
+ close(op.done)
+}
+
+func (c *configCache) clearWatchStateLocked(path string, err error) {
+ current := c.watches[path]
+ if current.pending {
+ c.completeWatchOperationLocked(current.pendingOp, err)
+ }
+ c.setWatchStateLocked(path, configWatchState{})
+}
+
+func (c *configCache) setWatch(path string, watchState configWatchState) bool {
+ pathLock := c.pathLock(path)
+ pathLock.Lock()
+ defer pathLock.Unlock()
+
+ c.stateLock.Lock()
+ defer c.stateLock.Unlock()
+ return c.setWatchStateLocked(path, watchState)
+}
+
+// setWatchStateLocked is the single accounting point for auto watch capacity.
+// The caller must hold stateLock while replacing a path's watch state.
+func (c *configCache) setWatchStateLocked(path string, watchState
configWatchState) bool {
+ current := c.watches[path]
+ if !watchState.pending {
+ watchState.pendingOp = nil
+ }
+ if watchState.auto && !c.enabled() {
+ return false
+ }
+ if watchState.holdsAutoSlot() && !current.holdsAutoSlot() &&
+ c.autoWatchCount+c.autoWatchReservations >= maxAutoWatches {
+ return false
+ }
+ if current.holdsAutoWatch() {
+ c.autoWatchCount--
+ }
+ if current.holdsAutoReservation() {
+ c.autoWatchReservations--
+ }
+ if !watchState.tracked() {
+ delete(c.watches, path)
+ return true
+ }
+ c.watches[path] = watchState
+ if watchState.holdsAutoWatch() {
+ c.autoWatchCount++
+ }
+ if watchState.holdsAutoReservation() {
+ c.autoWatchReservations++
+ }
+ return true
+}
+
+func (c *configCache) finishWatchRegistration(path string, generation, token
uint64, registration watchRegistration) bool {
+ pathLock := c.pathLock(path)
+ pathLock.Lock()
+ defer pathLock.Unlock()
+ return c.finishWatchRegistrationLocked(path, generation, token,
registration)
+}
+
+// finishWatchRegistrationLocked commits a pending watch only when its token,
+// cache generation, and ZooKeeper session still match the current state.
+func (c *configCache) finishWatchRegistrationLocked(path string, generation,
token uint64, registration watchRegistration) bool {
+ sessionID, active := registration.resolve()
+ c.stateLock.Lock()
+ defer c.stateLock.Unlock()
+
+ watchState := c.watches[path]
+ if !watchState.pending {
+ return false
+ }
+ if (watchState.pendingOp != nil && watchState.pendingOp.token != token)
||
+ (watchState.pendingOp == nil && token != 0) {
+ return false
+ }
+ generationCurrent := c.generation == generation
+ sessionCurrent := c.sessionID == 0 ||
+ (sessionID != 0 && sessionID == c.sessionID)
+ if !active || !generationCurrent || !registration.sessionStable() ||
!sessionCurrent {
+ c.clearWatchStateLocked(path, errWatchRegistrationStale)
+ return false
+ }
+ op := watchState.pendingOp
+ watchState.registered = true
+ watchState.pending = false
+ watchState.sessionID = sessionID
+ watchState.pendingOp = nil
+ stored := c.setWatchStateLocked(path, watchState)
+ if stored {
+ c.completeWatchOperationLocked(op, nil)
+ }
+ return stored
+}
+
+func (c *configCache) ensureBusinessWatch(
+ path string,
+ register func() (watchRegistration, error),
+) error {
+ return c.ensureBusinessWatchWithRetry(path, register, 1)
+}
+
+func (c *configCache) ensureBusinessWatchWithRetry(
+ path string,
+ register func() (watchRegistration, error),
+ maxRetries int,
+) error {
+ return c.ensureBusinessWatchWithRetryIf(path, register, maxRetries, nil)
+}
+
+func (c *configCache) ensureBusinessWatchWithRetryIf(
+ path string,
+ register func() (watchRegistration, error),
+ maxRetries int,
+ stillNeeded func() bool,
+) error {
+ for attempt := 0; ; attempt++ {
+ if stillNeeded != nil && !stillNeeded() {
+ return errBusinessWatchCanceled
+ }
+ err := c.ensureBusinessWatchOnce(path, register)
+ if err == nil && stillNeeded != nil && !stillNeeded() {
+ return errBusinessWatchCanceled
+ }
+ if !errors.Is(err, errWatchRegistrationStale) || attempt >=
maxRetries {
+ return err
+ }
+ }
+}
+
+func (c *configCache) ensureBusinessWatchOnce(
+ path string,
+ register func() (watchRegistration, error),
+) error {
+ pathLock := c.pathLock(path)
+ pathLock.Lock()
+ c.stateLock.Lock()
+ watchState := c.watches[path]
+ if watchState.registered {
+ watchState.auto = false
+ watchState.retired = false
+ c.setWatchStateLocked(path, watchState)
+ c.stateLock.Unlock()
+ pathLock.Unlock()
+ return nil
+ }
+ if watchState.pending {
+ watchState.auto = false
+ watchState.retired = false
+ c.setWatchStateLocked(path, watchState)
+ op := watchState.pendingOp
+ c.stateLock.Unlock()
+ pathLock.Unlock()
+ if op == nil {
+ return nil
+ }
+ <-op.done
+ return op.err
+ }
+ generation := c.generation
+ sessionID := c.sessionID
+ op := c.newWatchOperationLocked()
+ c.setWatchStateLocked(path, configWatchState{
+ pending: true,
+ sessionID: sessionID,
+ pendingOp: op,
+ })
+ c.stateLock.Unlock()
+ pathLock.Unlock()
+
+ registration, err := register()
+ pathLock.Lock()
+ defer pathLock.Unlock()
+ if err != nil {
+ c.stateLock.Lock()
+ current := c.watches[path]
+ if current.pendingOp == op {
+ c.clearWatchStateLocked(path, err)
+ } else {
+ c.completeWatchOperationLocked(op, err)
+ }
+ c.stateLock.Unlock()
+ return err
+ }
+ if c.finishWatchRegistrationLocked(path, generation, op.token,
registration) {
+ return nil
+ }
+ c.stateLock.Lock()
+ c.completeWatchOperationLocked(op, errWatchRegistrationStale)
+ c.stateLock.Unlock()
+ return errWatchRegistrationStale
+}
+
+func (c *configCache) releaseBusinessWatchLocked(path string) {
+ c.stateLock.Lock()
+ defer c.stateLock.Unlock()
+ watchState, ok := c.watches[path]
+ if !ok || watchState.auto {
+ return
+ }
+
+ autoWatchState := watchState
+ autoWatchState.auto = true
+ autoWatchState.retired = false
+ if c.setWatchStateLocked(path, autoWatchState) {
+ return
+ }
+ watchState.retired = true
+ c.setWatchStateLocked(path, watchState)
+}
+
+func (c *configCache) beginWatchRenewalLocked(path string, hasListeners bool)
(uint64, configWatchState, bool) {
+ c.stateLock.Lock()
+ defer c.stateLock.Unlock()
+ generation := c.generation
+ watchState, ok := c.watches[path]
+ if !ok {
+ if !hasListeners {
+ return generation, watchState, false
+ }
+ c.setWatchStateLocked(path, configWatchState{
+ pending: true,
+ sessionID: c.sessionID,
+ pendingOp: c.newWatchOperationLocked(),
+ })
+ return generation, c.watches[path], true
+ }
+ if watchState.pending {
+ if hasListeners {
+ watchState.auto = false
+ watchState.retired = false
+ c.setWatchStateLocked(path, watchState)
+ }
+ return generation, c.watches[path], false
+ }
+ if watchState.retired && !hasListeners {
+ c.setWatchStateLocked(path, configWatchState{})
+ return generation, configWatchState{}, false
+ }
+ if hasListeners {
+ watchState.auto = false
+ watchState.retired = false
+ } else if !watchState.auto {
+ c.setWatchStateLocked(path, configWatchState{})
+ return generation, configWatchState{}, false
+ }
+ watchState.registered = false
+ watchState.pending = true
+ watchState.pendingOp = c.newWatchOperationLocked()
+ c.setWatchStateLocked(path, watchState)
+ return generation, c.watches[path], true
+}
+
+func (c *configCache) cancelPendingLocked(path string) {
+ c.stateLock.Lock()
+ defer c.stateLock.Unlock()
+ if c.watches[path].pending {
+ c.clearWatchStateLocked(path, errWatchRegistrationStale)
+ }
+}
+
+// reset starts a new cache generation and clears cached values. Registered
+// watches from the same session are preserved because the ZooKeeper client
+// restores them on reconnect. Pending registrations remain for their
completion
+// path to reject by generation, while registered watches from older sessions
are
+// removed.
+func (c *configCache) reset(sessionID int64) {
+ c.stateLock.Lock()
+ defer c.stateLock.Unlock()
+
+ c.generation++
+ c.sessionID = sessionID
+ c.entries.Purge()
+ for path, watchState := range c.watches {
+ if watchState.pending || watchState.sessionID == sessionID {
+ continue
+ }
+ c.setWatchStateLocked(path, configWatchState{})
+ }
+}
+
+func (c *configCache) pathLock(path string) *sync.Mutex {
+ var hash uint32 = 2166136261
+ for i := 0; i < len(path); i++ {
+ hash ^= uint32(path[i])
+ hash *= 16777619
+ }
+ return &c.pathLocks[hash%pathLockShardCount]
+}
diff --git a/config_center/zookeeper/config_cache_test.go
b/config_center/zookeeper/config_cache_test.go
new file mode 100644
index 000000000..297655d4a
--- /dev/null
+++ b/config_center/zookeeper/config_cache_test.go
@@ -0,0 +1,572 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package zookeeper
+
+import (
+ "fmt"
+ "sync"
+ "sync/atomic"
+ "testing"
+ "time"
+)
+
+import (
+ "github.com/go-zookeeper/zk"
+
+ "github.com/stretchr/testify/require"
+)
+
+func newTestWatchRegistration(sessionID int64) watchRegistration {
+ return watchRegistration{
+ events: make(chan zk.Event, 1),
+ beforeSessionID: sessionID,
+ afterSessionID: sessionID,
+ }
+}
+
+func TestConfigCacheLoadAndExpiry(t *testing.T) {
+ cache := newConfigCache(20 * time.Millisecond)
+ var loads atomic.Int32
+ loader := func(registerWatch bool) (configCacheEntry,
watchRegistration, error) {
+ count := loads.Add(1)
+ if registerWatch {
+ return configCacheEntry{content: string(rune('0' +
count)), exists: true},
+ newTestWatchRegistration(1), nil
+ }
+ return configCacheEntry{content: string(rune('0' + count)),
exists: true}, watchRegistration{}, nil
+ }
+
+ first, err := cache.load("/path", loader)
+ require.NoError(t, err)
+ second, err := cache.load("/path", loader)
+ require.NoError(t, err)
+ require.Equal(t, first.content, second.content)
+ require.Equal(t, int32(1), loads.Load())
+
+ require.Eventually(t, func() bool {
+ entry, loadErr := cache.load("/path", loader)
+ return loadErr == nil && entry.content == "2"
+ }, time.Second, 5*time.Millisecond)
+ require.Equal(t, int32(2), loads.Load())
+}
+
+func TestConfigCacheUsesFixedPathLockShards(t *testing.T) {
+ cache := newConfigCache(time.Minute)
+ locks := make(map[*sync.Mutex]struct{})
+ pathLock := cache.pathLock("/path")
+
+ for i := range 4096 {
+ locks[cache.pathLock(fmt.Sprintf("/path/%d", i))] = struct{}{}
+ }
+
+ require.Same(t, pathLock, cache.pathLock("/path"))
+ require.LessOrEqual(t, len(locks), pathLockShardCount)
+}
+
+func TestConfigCacheBoundsEntriesUnderKeyChurn(t *testing.T) {
+ cache := newConfigCache(time.Minute)
+
+ for i := range 4096 {
+ cache.store(fmt.Sprintf("/path/%d", i), configCacheEntry{
+ content: fmt.Sprintf("value-%d", i),
+ exists: true,
+ })
+ }
+
+ require.Equal(t, maxCacheEntries, cache.entries.Len())
+}
+
+func TestConfigCacheStoresMissingEntry(t *testing.T) {
+ cache := newConfigCache(time.Minute)
+ cache.store("/missing", configCacheEntry{exists: false})
+
+ require.Equal(t, 1, cache.entries.Len())
+ entry, ok := cache.getFresh("/missing")
+ require.True(t, ok)
+ require.False(t, entry.exists)
+}
+
+func TestConfigCacheEvictsLeastRecentlyUsed(t *testing.T) {
+ cache := newConfigCache(time.Minute)
+ for i := range maxCacheEntries {
+ cache.store(fmt.Sprintf("/path/%d", i),
configCacheEntry{exists: true})
+ }
+
+ _, ok := cache.getFresh("/path/0")
+ require.True(t, ok)
+ cache.store("/new", configCacheEntry{exists: true})
+
+ _, ok = cache.getFresh("/path/1")
+ require.False(t, ok)
+ _, ok = cache.getFresh("/path/0")
+ require.True(t, ok)
+}
+
+func TestConfigCacheRemovesExpiredEntryOnAccess(t *testing.T) {
+ cache := newConfigCache(10 * time.Millisecond)
+ cache.store("/path", configCacheEntry{exists: true})
+ require.Equal(t, 1, cache.entries.Len())
+ time.Sleep(20 * time.Millisecond)
+ require.Equal(t, 1, cache.entries.Len())
+
+ _, ok := cache.getFresh("/path")
+ require.False(t, ok)
+ require.Zero(t, cache.entries.Len())
+}
+
+func TestConfigCacheConcurrentLoadsRemainBounded(t *testing.T) {
+ cache := newConfigCache(time.Minute)
+ var wg sync.WaitGroup
+ errs := make(chan error, 4096)
+
+ for i := range 4096 {
+ path := fmt.Sprintf("/path/%d", i)
+ wg.Go(func() {
+ _, err := cache.load(path, func(registerWatch bool)
(configCacheEntry, watchRegistration, error) {
+ if registerWatch {
+ return configCacheEntry{exists: true},
newTestWatchRegistration(1), nil
+ }
+ return configCacheEntry{exists: true},
watchRegistration{}, nil
+ })
+ errs <- err
+ })
+ }
+ wg.Wait()
+ close(errs)
+ for err := range errs {
+ require.NoError(t, err)
+ }
+
+ require.Equal(t, maxCacheEntries, cache.entries.Len())
+}
+
+func TestConfigCacheConcurrentAutoWatchRegistrationsRemainBounded(t
*testing.T) {
+ cache := newConfigCache(time.Minute)
+ var registrations atomic.Int32
+ var wg sync.WaitGroup
+ errs := make(chan error, 4096)
+
+ for i := range 4096 {
+ path := fmt.Sprintf("/watch/%d", i)
+ wg.Go(func() {
+ _, err := cache.load(path, func(registerWatch bool)
(configCacheEntry, watchRegistration, error) {
+ if registerWatch {
+ registrations.Add(1)
+ return configCacheEntry{exists: true},
newTestWatchRegistration(1), nil
+ }
+ return configCacheEntry{exists: true},
watchRegistration{}, nil
+ })
+ errs <- err
+ })
+ }
+ wg.Wait()
+ close(errs)
+ for err := range errs {
+ require.NoError(t, err)
+ }
+
+ require.Equal(t, int32(maxAutoWatches), registrations.Load())
+ require.Equal(t, maxAutoWatches, cache.autoWatchCount)
+ require.Zero(t, cache.autoWatchReservations)
+}
+
+func TestConfigCacheWatchUpdateWinsOverLoad(t *testing.T) {
+ cache := newConfigCache(time.Minute)
+ loadStarted := make(chan struct{})
+ releaseLoad := make(chan struct{})
+ loadDone := make(chan struct{})
+ go func() {
+ defer close(loadDone)
+ _, _ = cache.load("/path", func(registerWatch bool)
(configCacheEntry, watchRegistration, error) {
+ close(loadStarted)
+ <-releaseLoad
+ if registerWatch {
+ return configCacheEntry{content: "old", exists:
true}, newTestWatchRegistration(1), nil
+ }
+ return configCacheEntry{content: "old", exists: true},
watchRegistration{}, nil
+ })
+ }()
+
+ <-loadStarted
+ updateDone := make(chan struct{})
+ go func() {
+ defer close(updateDone)
+ cache.store("/path", configCacheEntry{content: "new", exists:
true})
+ }()
+ close(releaseLoad)
+ <-loadDone
+ <-updateDone
+
+ entry, ok := cache.getFresh("/path")
+ require.True(t, ok)
+ require.Equal(t, "new", entry.content)
+}
+
+func TestConfigCacheResetDiscardsInFlightLoad(t *testing.T) {
+ cache := newConfigCache(time.Minute)
+ loadStarted := make(chan struct{})
+ releaseLoad := make(chan struct{})
+ type loadResult struct {
+ entry configCacheEntry
+ err error
+ }
+ result := make(chan loadResult, 1)
+ var loads atomic.Int32
+
+ go func() {
+ entry, err := cache.load("/path", func(bool) (configCacheEntry,
watchRegistration, error) {
+ if loads.Add(1) == 1 {
+ close(loadStarted)
+ <-releaseLoad
+ return configCacheEntry{content: "old", exists:
true}, newTestWatchRegistration(1), nil
+ }
+ return configCacheEntry{content: "new", exists: true},
newTestWatchRegistration(2), nil
+ })
+ result <- loadResult{entry: entry, err: err}
+ }()
+
+ <-loadStarted
+ cache.reset(2)
+ _, ok := cache.getFresh("/path")
+ require.False(t, ok)
+ _, watchState := cache.snapshot("/path")
+ require.True(t, watchState.pending)
+ require.Equal(t, 1, cache.autoWatchReservations)
+ close(releaseLoad)
+
+ load := <-result
+ require.NoError(t, load.err)
+ require.Equal(t, "new", load.entry.content)
+ require.Equal(t, int32(2), loads.Load())
+ entry, ok := cache.getFresh("/path")
+ require.True(t, ok)
+ require.Equal(t, "new", entry.content)
+ _, watchState = cache.snapshot("/path")
+ require.True(t, watchState.registered)
+ require.Equal(t, int64(2), watchState.sessionID)
+}
+
+func TestConfigCacheRejectsInFlightWatchAfterGenerationReset(t *testing.T) {
+ cache := newConfigCache(time.Minute)
+ loadStarted := make(chan struct{})
+ releaseLoad := make(chan struct{})
+ result := make(chan error, 1)
+ decisions := make(chan bool, 2)
+ var loads atomic.Int32
+ var registrations atomic.Int32
+
+ go func() {
+ _, err := cache.load("/path", func(registerWatch bool)
(configCacheEntry, watchRegistration, error) {
+ decisions <- registerWatch
+ if loads.Add(1) == 1 {
+ registrations.Add(1)
+ close(loadStarted)
+ <-releaseLoad
+ return configCacheEntry{content: "old", exists:
true}, newTestWatchRegistration(1), nil
+ }
+ registrations.Add(1)
+ return configCacheEntry{content: "new", exists: true},
newTestWatchRegistration(1), nil
+ })
+ result <- err
+ }()
+
+ <-loadStarted
+ cache.reset(1)
+ close(releaseLoad)
+
+ require.NoError(t, <-result)
+ require.Equal(t, int32(2), loads.Load())
+ require.Equal(t, int32(2), registrations.Load())
+ require.True(t, <-decisions)
+ require.True(t, <-decisions)
+ entry, ok := cache.getFresh("/path")
+ require.True(t, ok)
+ require.Equal(t, "new", entry.content)
+ _, watchState := cache.snapshot("/path")
+ require.True(t, watchState.registered)
+ require.Equal(t, int64(1), watchState.sessionID)
+}
+
+func TestConfigCacheResetPreservesOnlyCurrentSessionWatches(t *testing.T) {
+ cache := newConfigCache(time.Minute)
+ cache.store("/entry", configCacheEntry{content: "value", exists: true})
+ require.True(t, cache.setWatch("/active", configWatchState{
+ registered: true,
+ auto: true,
+ sessionID: 2,
+ }))
+ require.True(t, cache.setWatch("/retired", configWatchState{
+ registered: true,
+ retired: true,
+ sessionID: 2,
+ }))
+ require.True(t, cache.setWatch("/stale", configWatchState{
+ registered: true,
+ auto: true,
+ sessionID: 1,
+ }))
+ require.True(t, cache.setWatch("/pending", configWatchState{
+ pending: true,
+ auto: true,
+ sessionID: 1,
+ }))
+
+ cache.reset(2)
+
+ _, ok := cache.getFresh("/entry")
+ require.False(t, ok)
+ _, active := cache.snapshot("/active")
+ require.True(t, active.registered)
+ _, retired := cache.snapshot("/retired")
+ require.True(t, retired.registered)
+ require.True(t, retired.retired)
+ _, stale := cache.snapshot("/stale")
+ require.False(t, stale.tracked())
+ _, pending := cache.snapshot("/pending")
+ require.True(t, pending.pending)
+ require.Equal(t, 1, cache.autoWatchCount)
+ require.Equal(t, 1, cache.autoWatchReservations)
+}
+
+func TestWatchRegistrationResolveAcrossSession(t *testing.T) {
+ t.Run("current session watch remains active", func(t *testing.T) {
+ registration := watchRegistration{
+ events: make(chan zk.Event, 1),
+ beforeSessionID: 1,
+ afterSessionID: 2,
+ }
+
+ sessionID, active := registration.resolve()
+ require.True(t, active)
+ require.Equal(t, int64(2), sessionID)
+ })
+
+ t.Run("invalidated watch is discarded", func(t *testing.T) {
+ events := make(chan zk.Event, 1)
+ events <- zk.Event{Type: zk.EventNotWatching}
+ close(events)
+ registration := watchRegistration{
+ events: events,
+ beforeSessionID: 1,
+ afterSessionID: 2,
+ }
+
+ sessionID, active := registration.resolve()
+ require.False(t, active)
+ require.Zero(t, sessionID)
+ })
+}
+
+func TestWatchRegistrationSessionStabilityIncludesFollowUpRead(t *testing.T) {
+ registration := watchRegistration{
+ events: make(chan zk.Event, 1),
+ beforeSessionID: 1,
+ afterSessionID: 1,
+ readBeforeSessionID: 2,
+ readAfterSessionID: 2,
+ }
+ require.False(t, registration.sessionStable())
+
+ registration.readBeforeSessionID = 1
+ registration.readAfterSessionID = 1
+ require.True(t, registration.sessionStable())
+}
+
+func TestConfigCacheResetDiscardsInFlightBusinessWatch(t *testing.T) {
+ cache := newConfigCache(time.Minute)
+ registerStarted := make(chan struct{})
+ releaseRegister := make(chan struct{})
+ result := make(chan error, 1)
+ var registrations atomic.Int32
+
+ go func() {
+ result <- cache.ensureBusinessWatch("/path", func()
(watchRegistration, error) {
+ if registrations.Add(1) == 1 {
+ close(registerStarted)
+ <-releaseRegister
+ return newTestWatchRegistration(1), nil
+ }
+ return newTestWatchRegistration(2), nil
+ })
+ }()
+
+ <-registerStarted
+ cache.reset(2)
+ _, pendingState := cache.snapshot("/path")
+ require.True(t, pendingState.pending)
+ close(releaseRegister)
+
+ require.NoError(t, <-result)
+ require.Equal(t, int32(2), registrations.Load())
+ _, watchState := cache.snapshot("/path")
+ require.True(t, watchState.registered)
+ require.False(t, watchState.auto)
+ require.Equal(t, int64(2), watchState.sessionID)
+}
+
+func
TestConfigCacheConcurrentBusinessWatchRegistrationSharesPendingOperation(t
*testing.T) {
+ cache := newConfigCache(time.Minute)
+ registerStarted := make(chan struct{})
+ releaseRegister := make(chan struct{})
+ var registrations atomic.Int32
+ register := func() (watchRegistration, error) {
+ if registrations.Add(1) == 1 {
+ close(registerStarted)
+ <-releaseRegister
+ }
+ return newTestWatchRegistration(1), nil
+ }
+
+ firstResult := make(chan error, 1)
+ go func() { firstResult <- cache.ensureBusinessWatch("/path", register)
}()
+ <-registerStarted
+
+ secondResult := make(chan error, 1)
+ go func() { secondResult <- cache.ensureBusinessWatch("/path",
register) }()
+ time.Sleep(10 * time.Millisecond)
+ require.Equal(t, int32(1), registrations.Load())
+
+ close(releaseRegister)
+ require.NoError(t, <-firstResult)
+ require.NoError(t, <-secondResult)
+ require.Equal(t, int32(1), registrations.Load())
+ _, watchState := cache.snapshot("/path")
+ require.True(t, watchState.registered)
+ require.False(t, watchState.auto)
+}
+
+func TestConfigCacheRegistrationRequiresCurrentGenerationAndSession(t
*testing.T) {
+ cache := newConfigCache(time.Minute)
+ cache.reset(1)
+
+ generation, registerWatch, token := cache.prepareLoad("/path")
+ require.True(t, registerWatch)
+
+ cache.reset(1)
+ require.False(t, cache.finishWatchRegistration("/path", generation,
token, newTestWatchRegistration(1)))
+ _, watchState := cache.snapshot("/path")
+ require.False(t, watchState.tracked())
+
+ generation, registerWatch, token = cache.prepareLoad("/path")
+ require.True(t, registerWatch)
+ cache.stateLock.Lock()
+ cache.sessionID = 2
+ cache.stateLock.Unlock()
+ require.False(t, cache.finishWatchRegistration("/path", generation,
token, newTestWatchRegistration(1)))
+ _, watchState = cache.snapshot("/path")
+ require.False(t, watchState.tracked())
+}
+
+func TestConfigCacheLoadRetryIsBoundedWhenGenerationKeepsChanging(t
*testing.T) {
+ cache := newConfigCache(time.Minute)
+ var loads atomic.Int32
+ started := time.Now()
+ _, err := cache.load("/path", func(registerWatch bool)
(configCacheEntry, watchRegistration, error) {
+ loads.Add(1)
+ sessionID := time.Now().UnixNano()
+ cache.reset(sessionID)
+ if registerWatch {
+ return configCacheEntry{content: "stale", exists:
true}, newTestWatchRegistration(sessionID), nil
+ }
+ return configCacheEntry{content: "stale", exists: true},
watchRegistration{}, nil
+ })
+
+ require.ErrorIs(t, err, errWatchRegistrationStale)
+ require.LessOrEqual(t, loads.Load(), int32(2*(maxCacheLoadRetries+1)+1))
+ require.Less(t, time.Since(started), time.Second)
+}
+
+func TestConfigCacheBusinessWatchRegistrationRetryIsBounded(t *testing.T) {
+ cache := newConfigCache(time.Minute)
+ var registrations atomic.Int32
+ invalidRegistration := func() (watchRegistration, error) {
+ registrations.Add(1)
+ events := make(chan zk.Event, 1)
+ events <- zk.Event{Type: zk.EventNotWatching}
+ close(events)
+ return watchRegistration{
+ events: events,
+ beforeSessionID: 1,
+ afterSessionID: 2,
+ }, nil
+ }
+
+ err := cache.ensureBusinessWatchWithRetry("/path", invalidRegistration,
1)
+ require.ErrorIs(t, err, errWatchRegistrationStale)
+ require.Equal(t, int32(2), registrations.Load())
+ _, watchState := cache.snapshot("/path")
+ require.False(t, watchState.tracked())
+}
+
+func TestConfigCacheLoadRejectsStaleRegistrationBeforeStoringEntry(t
*testing.T) {
+ cache := newConfigCache(time.Minute)
+ var registrations atomic.Int32
+
+ loader := func(registerWatch bool) (configCacheEntry,
watchRegistration, error) {
+ if !registerWatch {
+ return configCacheEntry{content: "fallback", exists:
true}, watchRegistration{}, nil
+ }
+ if registrations.Add(1) == 1 {
+ events := make(chan zk.Event, 1)
+ events <- zk.Event{Type: zk.EventNotWatching}
+ close(events)
+ return configCacheEntry{content: "stale", exists:
true}, watchRegistration{
+ events: events,
+ beforeSessionID: 1,
+ afterSessionID: 2,
+ }, nil
+ }
+ return configCacheEntry{content: "current", exists: true},
newTestWatchRegistration(2), nil
+ }
+
+ entry, err := cache.load("/path", loader)
+ require.NoError(t, err)
+ require.Equal(t, "current", entry.content)
+ require.Equal(t, int32(2), registrations.Load())
+ cached, ok := cache.getFresh("/path")
+ require.True(t, ok)
+ require.Equal(t, "current", cached.content)
+}
+
+func TestConfigCacheLoadFallsBackAfterRepeatedStaleRegistrations(t *testing.T)
{
+ cache := newConfigCache(time.Minute)
+ var registrations atomic.Int32
+ var fallbacks atomic.Int32
+
+ loader := func(registerWatch bool) (configCacheEntry,
watchRegistration, error) {
+ if !registerWatch {
+ fallbacks.Add(1)
+ return configCacheEntry{content: "fallback", exists:
true}, watchRegistration{}, nil
+ }
+ registrations.Add(1)
+ events := make(chan zk.Event, 1)
+ events <- zk.Event{Type: zk.EventNotWatching}
+ close(events)
+ return configCacheEntry{content: "stale", exists: true},
watchRegistration{
+ events: events,
+ beforeSessionID: 1,
+ afterSessionID: 2,
+ }, nil
+ }
+
+ entry, err := cache.load("/path", loader)
+ require.NoError(t, err)
+ require.Equal(t, "fallback", entry.content)
+ require.Equal(t, int32(2), registrations.Load())
+ require.Equal(t, int32(1), fallbacks.Load())
+}
diff --git a/config_center/zookeeper/impl.go b/config_center/zookeeper/impl.go
index fdf4aa383..ff554bcb9 100644
--- a/config_center/zookeeper/impl.go
+++ b/config_center/zookeeper/impl.go
@@ -24,6 +24,7 @@ import (
"strconv"
"strings"
"sync"
+ "time"
)
import (
@@ -45,7 +46,10 @@ import (
)
const (
- pathSeparator = "/"
+ pathSeparator = "/"
+ defaultConfigCacheTTL = 30 * time.Second
+ minConfigCacheTTL = time.Second
+ maxConfigCacheTTL = 10 * time.Minute
)
type zookeeperDynamicConfiguration struct {
@@ -60,6 +64,7 @@ type zookeeperDynamicConfiguration struct {
// listenerLock sync.Mutex
listener *zookeeper.ZkEventListener
cacheListener *CacheListener
+ cache configCache
parser parser.ConfigurationParser
base64Enabled bool
@@ -67,20 +72,25 @@ type zookeeperDynamicConfiguration struct {
func newZookeeperDynamicConfiguration(url *common.URL)
(*zookeeperDynamicConfiguration, error) {
rootPath := url.GetParam(constant.ConfigRootPathParamKey,
"/dubbo/config")
+ cacheTTL, err :=
parseConfigCacheTTL(url.GetParam(constant.ConfigCacheTTLKey, ""))
+ if err != nil {
+ return nil, err
+ }
c := &zookeeperDynamicConfiguration{
url: url,
rootPath: rootPath,
+ cache: newConfigCache(cacheTTL),
}
logger.Infof("[ConfigCenter][Zookeeper] new Zookeeper ConfigCenter with
Configuration, zkConfig=%v url=%v", c, c.GetURL())
if v := url.GetParam("base64", ""); v != "" {
- base64Enabled, err := strconv.ParseBool(v)
- if err != nil {
- panic("value of base64 must be bool, error=" +
err.Error())
+ base64Enabled, parseErr := strconv.ParseBool(v)
+ if parseErr != nil {
+ panic("value of base64 must be bool, error=" +
parseErr.Error())
}
c.base64Enabled = base64Enabled
}
- err := zookeeper.ValidateZookeeperClient(c, url.Location)
+ err = zookeeper.ValidateZookeeperClient(c, url.Location)
if err != nil {
logger.Errorf("[ConfigCenter][Zookeeper] zookeeper client start
error, err=%v", err)
return nil, err
@@ -96,16 +106,14 @@ func newZookeeperDynamicConfiguration(url *common.URL)
(*zookeeperDynamicConfigu
// Start listener
c.listener = zookeeper.NewZkEventListener(c.client)
- c.cacheListener = NewCacheListener(c.rootPath, c.listener)
+ c.cacheListener = newCacheListener(c.rootPath, c.listener, &c.cache)
c.listener.ListenConfigurationEvent(c.rootPath, c.cacheListener)
return c, nil
}
// AddListener add listener for key
-// TODO this method should has a parameter 'group', and it does not now, so we
should concat group and key with '/' manually
func (c *zookeeperDynamicConfiguration) AddListener(key string, listener
config_center.ConfigurationListener, options ...config_center.Option) {
- key =
strings.Join([]string{c.GetURL().GetParam(constant.ConfigNamespaceKey,
config_center.DefaultGroup), key}, "/")
- qualifiedKey := buildPath(c.rootPath, key)
+ qualifiedKey := c.getPropertiesPath(key, options...)
c.cacheListener.AddListener(qualifiedKey, listener)
}
@@ -119,11 +127,34 @@ func buildPath(rootPath, subPath string) string {
return path.Clean(fullPath)
}
-func (c *zookeeperDynamicConfiguration) RemoveListener(key string, listener
config_center.ConfigurationListener, opions ...config_center.Option) {
- c.cacheListener.RemoveListener(key, listener)
+func (c *zookeeperDynamicConfiguration) RemoveListener(key string, listener
config_center.ConfigurationListener, options ...config_center.Option) {
+ qualifiedKey := c.getPropertiesPath(key, options...)
+ c.cacheListener.RemoveListener(qualifiedKey, listener)
}
func (c *zookeeperDynamicConfiguration) GetProperties(key string, opts
...config_center.Option) (string, error) {
+ path := c.getPropertiesPath(key, opts...)
+ entry, err := c.cache.load(path, func(registerWatch bool)
(configCacheEntry, watchRegistration, error) {
+ return c.loadProperties(path, registerWatch)
+ })
+ if err != nil {
+ return "", err
+ }
+ if !entry.exists {
+ return "", nil
+ }
+ if !c.base64Enabled {
+ return entry.content, nil
+ }
+
+ decoded, err := base64.StdEncoding.DecodeString(entry.content)
+ if err != nil {
+ return "", perrors.WithStack(err)
+ }
+ return string(decoded), nil
+}
+
+func (c *zookeeperDynamicConfiguration) getPropertiesPath(key string, opts
...config_center.Option) string {
tmpOpts := config_center.NewOptions(opts...)
/**
* when group is not null, we are getting startup configs from Config
Center, for example:
@@ -134,23 +165,88 @@ func (c *zookeeperDynamicConfiguration) GetProperties(key
string, opts ...config
} else {
key = c.GetURL().GetParam(constant.ConfigNamespaceKey,
config_center.DefaultGroup) + "/" + key
}
- content, _, err := c.client.GetContent(c.rootPath + "/" + key)
- if errors.Is(err, zk.ErrNoNode) {
- logger.Warnf("[ConfigCenter][Zookeeper] query rule fail, key=%s
err=%v", key, err)
- return "", nil
- }
- if err != nil {
- return "", perrors.WithStack(err)
+ return buildPath(c.rootPath, key)
+}
+
+func (c *zookeeperDynamicConfiguration) loadProperties(path string,
registerWatch bool) (configCacheEntry, watchRegistration, error) {
+ if !c.cache.enabled() || !registerWatch {
+ beforeSessionID := c.client.Conn.SessionID()
+ content, _, err := c.client.GetContent(path)
+ afterSessionID := c.client.Conn.SessionID()
+ registration := watchRegistration{
+ beforeSessionID: beforeSessionID,
+ afterSessionID: afterSessionID,
+ }
+ if errors.Is(err, zk.ErrNoNode) {
+ logger.Warnf("[ConfigCenter][Zookeeper] query rule
fail, key=%s err=%v", path, err)
+ return configCacheEntry{exists: false}, registration,
nil
+ }
+ if err != nil {
+ return configCacheEntry{}, registration,
perrors.WithStack(err)
+ }
+ return configCacheEntry{content: string(content), exists:
true}, registration, nil
}
- if !c.base64Enabled {
- return string(content), nil
+
+ for {
+ beforeSessionID := c.client.Conn.SessionID()
+ content, _, events, err := c.client.Conn.GetW(path)
+ registration := watchRegistration{
+ events: events,
+ beforeSessionID: beforeSessionID,
+ afterSessionID: c.client.Conn.SessionID(),
+ }
+ if err == nil {
+ return configCacheEntry{content: string(content),
exists: true}, registration, nil
+ }
+ if !errors.Is(err, zk.ErrNoNode) {
+ return configCacheEntry{}, watchRegistration{},
perrors.WithStack(err)
+ }
+
+ beforeSessionID = c.client.Conn.SessionID()
+ exists, _, events, watchErr := c.client.Conn.ExistsW(path)
+ registration = watchRegistration{
+ events: events,
+ beforeSessionID: beforeSessionID,
+ afterSessionID: c.client.Conn.SessionID(),
+ }
+ if watchErr != nil {
+ return configCacheEntry{}, watchRegistration{},
perrors.WithStack(watchErr)
+ }
+ if !exists {
+ logger.Warnf("[ConfigCenter][Zookeeper] query rule
fail, key=%s err=%v", path, err)
+ return configCacheEntry{exists: false}, registration,
nil
+ }
+
+ readBeforeSessionID := c.client.Conn.SessionID()
+ content, _, getErr := c.client.Conn.Get(path)
+ readAfterSessionID := c.client.Conn.SessionID()
+ registration.readBeforeSessionID = readBeforeSessionID
+ registration.readAfterSessionID = readAfterSessionID
+ if errors.Is(getErr, zk.ErrNoNode) {
+ continue
+ }
+ if getErr != nil {
+ return configCacheEntry{}, registration,
perrors.WithStack(getErr)
+ }
+ return configCacheEntry{content: string(content), exists:
true}, registration, nil
}
+}
- decoded, err := base64.StdEncoding.DecodeString(string(content))
+func parseConfigCacheTTL(value string) (time.Duration, error) {
+ if value == "" {
+ return defaultConfigCacheTTL, nil
+ }
+ ttl, err := time.ParseDuration(value)
if err != nil {
- return "", perrors.WithStack(err)
+ return 0, perrors.Wrapf(err, "invalid %s value %q",
constant.ConfigCacheTTLKey, value)
}
- return string(decoded), nil
+ if ttl < 0 {
+ return 0, perrors.Errorf("%s must not be negative",
constant.ConfigCacheTTLKey)
+ }
+ if ttl > 0 && (ttl < minConfigCacheTTL || ttl > maxConfigCacheTTL) {
+ return 0, perrors.Errorf("%s must be 0 or between %s and %s",
constant.ConfigCacheTTLKey, minConfigCacheTTL, maxConfigCacheTTL)
+ }
+ return ttl, nil
}
// GetInternalProperty For zookeeper, getConfig and getConfigs have the same
meaning.
@@ -274,6 +370,14 @@ func (c *zookeeperDynamicConfiguration) closeConfigs() {
}
func (c *zookeeperDynamicConfiguration) RestartCallBack() bool {
+ var sessionID int64
+ if c.client != nil && c.client.Conn != nil {
+ sessionID = c.client.Conn.SessionID()
+ }
+ c.cache.reset(sessionID)
+ if c.cacheListener != nil {
+ c.cacheListener.restoreBusinessWatches()
+ }
return true
}
diff --git a/config_center/zookeeper/impl_test.go
b/config_center/zookeeper/impl_test.go
index af2f9dbc7..52644f4d8 100644
--- a/config_center/zookeeper/impl_test.go
+++ b/config_center/zookeeper/impl_test.go
@@ -19,9 +19,12 @@ package zookeeper
import (
"crypto/rand"
+ "encoding/base64"
"encoding/hex"
+ "fmt"
"os"
"testing"
+ "time"
)
import (
@@ -35,8 +38,17 @@ import (
import (
"dubbo.apache.org/dubbo-go/v3/common"
"dubbo.apache.org/dubbo-go/v3/config_center"
+ remotingzookeeper "dubbo.apache.org/dubbo-go/v3/remoting/zookeeper"
)
+type channelConfigListener struct {
+ events chan *config_center.ConfigChangeEvent
+}
+
+func (l *channelConfigListener) Process(event
*config_center.ConfigChangeEvent) {
+ l.events <- event
+}
+
// zkAddrEnvKey mirrors the environment variable gost's
// NewZookeeperClientFromEnv reads to locate a ZooKeeper server (see
// database/kv/zk/client.go in dubbogo/gost); it isn't exported there, so the
@@ -59,6 +71,17 @@ func failOrSkipZkUnavailable(t *testing.T, err error) {
t.Skipf("skip zk setup: %v", err)
}
+func newZookeeperTestClient(t *testing.T, name string)
(*gxzookeeper.ZookeeperClient, <-chan zk.Event) {
+ t.Helper()
+ client, events, err := gxzookeeper.NewZookeeperClientFromEnv(name,
5*time.Second)
+ if err != nil {
+ failOrSkipZkUnavailable(t, err)
+ return nil, nil
+ }
+ t.Cleanup(func() { client.Close() })
+ return client, events
+}
+
// newTestRoot returns a randomly named zookeeper root path unique to the
// calling test, and registers a t.Cleanup that recursively removes it via
// client once the test finishes, so runs stay isolated on a shared
@@ -193,6 +216,321 @@ func TestGetPropertiesWithZk(t *testing.T) {
require.Empty(t, empty)
}
+func TestLoadPropertiesRegistersWatchOnlyWhenInactive(t *testing.T) {
+ client, events := newZookeeperTestClient(t, "watch-selection")
+ root := newTestRoot(t, client)
+
+ cfg := &zookeeperDynamicConfiguration{
+ rootPath: root,
+ client: client,
+ url: mustURL(t, "registry://127.0.0.1:2181"),
+ cache: newConfigCache(time.Minute),
+ }
+ activePath := cfg.getPath("active", "group")
+ inactivePath := cfg.getPath("inactive", "group")
+ require.NoError(t, cfg.PublishConfig("active", "group", "v1"))
+ require.NoError(t, cfg.PublishConfig("inactive", "group", "v1"))
+
+ waitForEvent := func(path string, timeout time.Duration) bool {
+ timer := time.NewTimer(timeout)
+ defer timer.Stop()
+ for {
+ select {
+ case event := <-events:
+ if event.Path == path && event.Type ==
zk.EventNodeDataChanged {
+ return true
+ }
+ case <-timer.C:
+ return false
+ }
+ }
+ }
+
+ _, registration, err := cfg.loadProperties(activePath, false)
+ require.NoError(t, err)
+ require.Nil(t, registration.events)
+ _, stat, err := client.GetContent(activePath)
+ require.NoError(t, err)
+ _, err = client.SetContent(activePath, []byte("v2"), stat.Version)
+ require.NoError(t, err)
+ require.False(t, waitForEvent(activePath, time.Second))
+
+ _, registration, err = cfg.loadProperties(inactivePath, true)
+ require.NoError(t, err)
+ require.NotNil(t, registration.events)
+ _, stat, err = client.GetContent(inactivePath)
+ require.NoError(t, err)
+ _, err = client.SetContent(inactivePath, []byte("v2"), stat.Version)
+ require.NoError(t, err)
+ require.True(t, waitForEvent(inactivePath, time.Second))
+}
+
+func TestListenerUsesGroupOption(t *testing.T) {
+ client, _ := newZookeeperTestClient(t, "listener-group")
+ root := newTestRoot(t, client)
+
+ zkListener := remotingzookeeper.NewZkEventListener(client)
+ defer zkListener.Close()
+ cfg := &zookeeperDynamicConfiguration{
+ rootPath: root,
+ client: client,
+ url: mustURL(t, "registry://127.0.0.1:2181"),
+ cache: newConfigCache(time.Minute),
+ listener: zkListener,
+ }
+ cfg.cacheListener = newCacheListener(cfg.rootPath, zkListener,
&cfg.cache)
+ key := "app.properties"
+ group := "custom"
+ path := cfg.getPropertiesPath(key, config_center.WithGroup(group))
+ rec := &recListener{}
+
+ cfg.AddListener(key, rec, config_center.WithGroup(group))
+ _, ok := cfg.cacheListener.keyListeners.Load(path)
+ require.True(t, ok)
+
+ cfg.RemoveListener(key, rec, config_center.WithGroup(group))
+ _, ok = cfg.cacheListener.keyListeners.Load(path)
+ require.False(t, ok)
+}
+
+func TestGetPropertiesFallsBackToTTLAtAutoWatchLimit(t *testing.T) {
+ client, events := newZookeeperTestClient(t, "watch-limit")
+ root := newTestRoot(t, client)
+
+ cfg := &zookeeperDynamicConfiguration{
+ rootPath: root,
+ client: client,
+ url: mustURL(t, "registry://127.0.0.1:2181"),
+ cache: newConfigCache(time.Minute),
+ }
+ for i := range maxAutoWatches {
+ require.True(t, cfg.cache.setWatch(fmt.Sprintf("/watch/%d", i),
configWatchState{
+ registered: true,
+ auto: true,
+ sessionID: client.Conn.SessionID(),
+ }))
+ }
+
+ require.NoError(t, cfg.PublishConfig("fallback", "group", "v1"))
+ value, err := cfg.GetProperties("fallback",
config_center.WithGroup("group"))
+ require.NoError(t, err)
+ require.Equal(t, "v1", value)
+
+ path := cfg.getPath("fallback", "group")
+ _, watchState := cfg.cache.snapshot(path)
+ require.False(t, watchState.tracked())
+ require.Equal(t, maxAutoWatches, cfg.cache.autoWatchCount)
+ require.Zero(t, cfg.cache.autoWatchReservations)
+ _, stat, err := client.GetContent(path)
+ require.NoError(t, err)
+ _, err = client.SetContent(path, []byte("v2"), stat.Version)
+ require.NoError(t, err)
+
+ timer := time.NewTimer(100 * time.Millisecond)
+ defer timer.Stop()
+ for {
+ select {
+ case event := <-events:
+ if event.Path == path && event.Type ==
zk.EventNodeDataChanged {
+ t.Fatal("TTL fallback should not register an
auto watch")
+ }
+ case <-timer.C:
+ value, err = cfg.GetProperties("fallback",
config_center.WithGroup("group"))
+ require.NoError(t, err)
+ require.Equal(t, "v1", value)
+ return
+ }
+ }
+}
+
+func TestGetPropertiesCacheUpdatedByWatch(t *testing.T) {
+ client, _ := newZookeeperTestClient(t, "cache-watch")
+ go (&gxzookeeper.DefaultHandler{}).HandleZkEvent(client)
+ root := newTestRoot(t, client)
+
+ cfg := &zookeeperDynamicConfiguration{
+ rootPath: root,
+ client: client,
+ done: make(chan struct{}),
+ url: mustURL(t, "registry://127.0.0.1:2181"),
+ cache: newConfigCache(time.Minute),
+ }
+ cfg.listener = remotingzookeeper.NewZkEventListener(client)
+ cfg.cacheListener = newCacheListener(cfg.rootPath, cfg.listener,
&cfg.cache)
+ cfg.listener.ListenConfigurationEvent(cfg.rootPath, cfg.cacheListener)
+ defer cfg.listener.Close()
+
+ require.NoError(t, cfg.PublishConfig("file.properties", "grp", "v1"))
+ value, err := cfg.GetProperties("file.properties",
config_center.WithGroup("grp"))
+ require.NoError(t, err)
+ require.Equal(t, "v1", value)
+
+ // ListenConfigurationEvent registers asynchronously; wait before
triggering the watch.
+ time.Sleep(50 * time.Millisecond)
+ watchPath := cfg.getPropertiesPath("file.properties",
config_center.WithGroup("grp"))
+ _, stat, err := client.GetContent(watchPath)
+ require.NoError(t, err)
+ _, err = client.SetContent(watchPath, []byte("v2"), stat.Version)
+ require.NoError(t, err)
+
+ require.Eventually(t, func() bool {
+ value, getErr := cfg.GetProperties("file.properties",
config_center.WithGroup("grp"))
+ return getErr == nil && value == "v2"
+ }, time.Second, 10*time.Millisecond)
+
+ require.NoError(t, cfg.RemoveConfig("file.properties", "grp"))
+ require.Eventually(t, func() bool {
+ entry, ok := cfg.cache.getFresh(watchPath)
+ return ok && !entry.exists
+ }, time.Second, 10*time.Millisecond)
+}
+
+func TestGetPropertiesDecodesCachedBase64(t *testing.T) {
+ cfg := &zookeeperDynamicConfiguration{
+ rootPath: "/dubbo/config",
+ url: mustURL(t, "registry://127.0.0.1:2181"),
+ cache: newConfigCache(time.Minute),
+ base64Enabled: true,
+ }
+ path := cfg.getPropertiesPath("key", config_center.WithGroup("group"))
+ cfg.cache.store(path, configCacheEntry{
+ content: base64.StdEncoding.EncodeToString([]byte("value")),
+ exists: true,
+ })
+
+ value, err := cfg.GetProperties("key", config_center.WithGroup("group"))
+ require.NoError(t, err)
+ require.Equal(t, "value", value)
+}
+
+func TestRestartCallBackResetsCache(t *testing.T) {
+ client, _ := newZookeeperTestClient(t, "restart-watch-reset")
+
+ cfg := &zookeeperDynamicConfiguration{cache:
newConfigCache(time.Minute), client: client}
+ path := "/dubbo/config/group/key"
+ pendingPath := "/dubbo/config/group/pending"
+ _, _, _, err := client.Conn.ExistsW(path)
+ require.NoError(t, err)
+ cfg.cache.store(path, configCacheEntry{content: "value", exists: true})
+ cfg.cache.setWatch(path, configWatchState{
+ registered: true,
+ auto: true,
+ sessionID: client.Conn.SessionID(),
+ })
+ cfg.cache.setWatch(pendingPath, configWatchState{
+ auto: true,
+ pending: true,
+ sessionID: client.Conn.SessionID(),
+ })
+
+ require.True(t, cfg.RestartCallBack())
+ _, ok := cfg.cache.getFresh(path)
+ require.False(t, ok)
+ _, watchState := cfg.cache.snapshot(path)
+ require.True(t, watchState.registered)
+ _, pendingWatchState := cfg.cache.snapshot(pendingPath)
+ require.True(t, pendingWatchState.pending)
+ require.Equal(t, 1, cfg.cache.autoWatchCount)
+ require.Equal(t, 1, cfg.cache.autoWatchReservations)
+}
+
+func TestRestartCallBackRestoresBusinessListener(t *testing.T) {
+ client, _ := newZookeeperTestClient(t, "restart-business-watch")
+ go (&gxzookeeper.DefaultHandler{}).HandleZkEvent(client)
+ root := newTestRoot(t, client)
+
+ cfg := &zookeeperDynamicConfiguration{
+ rootPath: root,
+ client: client,
+ url: mustURL(t, "registry://127.0.0.1:2181"),
+ cache: newConfigCache(time.Minute),
+ }
+ cfg.listener = remotingzookeeper.NewZkEventListener(client)
+ cfg.cacheListener = newCacheListener(cfg.rootPath, cfg.listener,
&cfg.cache)
+ cfg.listener.ListenConfigurationEvent(cfg.rootPath, cfg.cacheListener)
+ defer cfg.listener.Close()
+
+ key := "app.properties"
+ group := "group"
+ path := cfg.getPropertiesPath(key, config_center.WithGroup(group))
+ recorder := &channelConfigListener{events: make(chan
*config_center.ConfigChangeEvent, 2)}
+ require.NoError(t, cfg.PublishConfig(key, group, "v1"))
+ time.Sleep(50 * time.Millisecond)
+ cfg.AddListener(key, recorder, config_center.WithGroup(group))
+ _, previousWatch := cfg.cache.snapshot(path)
+ require.True(t, previousWatch.registered)
+ require.False(t, previousWatch.auto)
+
+ require.True(t, cfg.RestartCallBack())
+ _, restoredWatch := cfg.cache.snapshot(path)
+ require.True(t, restoredWatch.registered)
+ require.Equal(t, previousWatch.sessionID, restoredWatch.sessionID)
+ require.False(t, restoredWatch.auto)
+
+ for _, value := range []string{"v2", "v3"} {
+ _, stat, getErr := client.GetContent(path)
+ require.NoError(t, getErr)
+ _, setErr := client.SetContent(path, []byte(value),
stat.Version)
+ require.NoError(t, setErr)
+
+ select {
+ case event := <-recorder.events:
+ require.Equal(t, key, event.Key)
+ require.Equal(t, value, event.Value)
+ case <-time.After(time.Second):
+ t.Fatalf("listener did not receive configuration value
%q", value)
+ }
+ }
+}
+
+func TestRestartCallBackRestoresBusinessListenerWhenCacheDisabled(t
*testing.T) {
+ client, _ := newZookeeperTestClient(t,
"restart-business-watch-cache-disabled")
+ go (&gxzookeeper.DefaultHandler{}).HandleZkEvent(client)
+ root := newTestRoot(t, client)
+
+ cfg := &zookeeperDynamicConfiguration{
+ rootPath: root,
+ client: client,
+ url: mustURL(t, "registry://127.0.0.1:2181"),
+ cache: newConfigCache(0),
+ }
+ cfg.listener = remotingzookeeper.NewZkEventListener(client)
+ cfg.cacheListener = newCacheListener(cfg.rootPath, cfg.listener,
&cfg.cache)
+ defer cfg.listener.Close()
+
+ key := "app.properties"
+ group := "group"
+ path := cfg.getPropertiesPath(key, config_center.WithGroup(group))
+ recorder := &channelConfigListener{events: make(chan
*config_center.ConfigChangeEvent, 2)}
+ require.NoError(t, cfg.PublishConfig(key, group, "v1"))
+ time.Sleep(50 * time.Millisecond)
+ cfg.listener.ListenConfigurationEvent(cfg.rootPath, cfg.cacheListener)
+ cfg.AddListener(key, recorder, config_center.WithGroup(group))
+ _, watchState := cfg.cache.snapshot(path)
+ require.True(t, watchState.registered)
+ watchState.sessionID--
+ require.True(t, cfg.cache.setWatch(path, watchState))
+ require.True(t, cfg.RestartCallBack())
+ _, restoredWatch := cfg.cache.snapshot(path)
+ require.True(t, restoredWatch.registered)
+ require.Equal(t, client.Conn.SessionID(), restoredWatch.sessionID)
+
+ for _, value := range []string{"v2", "v3"} {
+ _, stat, getErr := client.GetContent(path)
+ require.NoError(t, getErr)
+ _, setErr := client.SetContent(path, []byte(value),
stat.Version)
+ require.NoError(t, setErr)
+
+ select {
+ case event := <-recorder.events:
+ require.Equal(t, key, event.Key)
+ require.Equal(t, value, event.Value)
+ case <-time.After(time.Second):
+ t.Fatalf("listener did not receive configuration value
%q", value)
+ }
+ }
+}
+
func mustURL(t *testing.T, raw string) *common.URL {
t.Helper()
u, err := common.NewURL(raw)
diff --git a/config_center/zookeeper/listener.go
b/config_center/zookeeper/listener.go
index d59692210..07097b649 100644
--- a/config_center/zookeeper/listener.go
+++ b/config_center/zookeeper/listener.go
@@ -22,6 +22,12 @@ import (
"sync"
)
+import (
+ "github.com/dubbogo/gost/log/logger"
+
+ "github.com/go-zookeeper/zk"
+)
+
import (
"dubbo.apache.org/dubbo-go/v3/common/constant"
"dubbo.apache.org/dubbo-go/v3/config_center"
@@ -34,60 +40,349 @@ import (
// CacheListener defines keyListeners and rootPath
type CacheListener struct {
// key is zkNode Path and value is set of listeners
- keyListeners sync.Map
+ keyListeners sync.Map
+ // eventGeneration carries the cache epoch and watch ownership from
event
+ // consumption through watch renewal to the final cache update.
+ eventGeneration sync.Map
zkEventListener *zookeeper.ZkEventListener
rootPath string
+ cache *configCache
+}
+
+// watchEventState is the snapshot attached to one consumed watch event. It
+// prevents a delayed renewal or DataChange callback from updating a newer
+// cache generation or completing a different pending registration.
+type watchEventState struct {
+ generation uint64
+ sessionID int64
+ auto bool
+ token uint64
}
// NewCacheListener creates a new CacheListener
func NewCacheListener(rootPath string, listener *zookeeper.ZkEventListener)
*CacheListener {
- return &CacheListener{zkEventListener: listener, rootPath: rootPath}
+ return newCacheListener(rootPath, listener, nil)
+}
+
+func newCacheListener(rootPath string, listener *zookeeper.ZkEventListener,
cache *configCache) *CacheListener {
+ return &CacheListener{zkEventListener: listener, rootPath: rootPath,
cache: cache}
+}
+
+func (l *CacheListener) registerWatcher(key string) (watchRegistration, error)
{
+ conn := l.zkEventListener.Client.Conn
+ beforeSessionID := conn.SessionID()
+ _, _, events, err := conn.ExistsW(key)
+ return watchRegistration{
+ events: events,
+ beforeSessionID: beforeSessionID,
+ afterSessionID: conn.SessionID(),
+ }, err
}
// AddListener will add a listener if loaded
func (l *CacheListener) AddListener(key string, listener
config_center.ConfigurationListener) {
// FIXME do not use Client.ExistW, cause it has a bug(can not watch zk
node that do not exist)
- _, _, _, err := l.zkEventListener.Client.Conn.ExistsW(key)
+ l.addListenerWithRegister(key, listener, func() (watchRegistration,
error) {
+ return l.registerWatcher(key)
+ })
+}
+
+func (l *CacheListener) addListenerWithRegister(
+ key string,
+ listener config_center.ConfigurationListener,
+ register func() (watchRegistration, error),
+) {
+ if l.cache == nil {
+ if _, err := register(); err != nil {
+ return
+ }
+ l.storeListener(key, listener)
+ return
+ }
+
+ pathLock := l.cache.pathLock(key)
+ pathLock.Lock()
+ added := l.storeListener(key, listener)
+ pathLock.Unlock()
+ stillNeeded := func() bool {
+ pathLock.Lock()
+ defer pathLock.Unlock()
+ return l.hasListener(key, listener)
+ }
+ err := l.cache.ensureBusinessWatchWithRetryIf(key, register, 1,
stillNeeded)
+ pathLock.Lock()
+ defer pathLock.Unlock()
+ present := l.hasListener(key, listener)
+ if err != nil && added && present {
+ _, last := l.removeListener(key, listener)
+ if last {
+ l.cache.releaseBusinessWatchLocked(key)
+ }
+ return
+ }
+ if !present && !l.hasListeners(key) {
+ l.cache.releaseBusinessWatchLocked(key)
+ }
+}
+
+func (l *CacheListener) storeListener(key string, listener
config_center.ConfigurationListener) bool {
// reference from
https://stackoverflow.com/questions/34018908/golang-why-dont-we-have-a-set-datastructure
// make a map[your type]struct{} like set in java
- if err != nil {
+ listeners, loaded := l.keyListeners.LoadOrStore(key,
map[config_center.ConfigurationListener]struct{}{listener: {}})
+ if !loaded {
+ return true
+ }
+ listenerSet :=
listeners.(map[config_center.ConfigurationListener]struct{})
+ if _, exists := listenerSet[listener]; exists {
+ return false
+ }
+ listenerSet[listener] = struct{}{}
+ l.keyListeners.Store(key, listenerSet)
+ return true
+}
+
+func (l *CacheListener) restoreBusinessWatches() {
+ if l.cache == nil || l.zkEventListener == nil ||
+ l.zkEventListener.Client == nil ||
l.zkEventListener.Client.Conn == nil {
return
}
- listeners, loaded := l.keyListeners.LoadOrStore(key,
map[config_center.ConfigurationListener]struct{}{listener: {}})
- if loaded {
-
listeners.(map[config_center.ConfigurationListener]struct{})[listener] =
struct{}{}
- l.keyListeners.Store(key, listeners)
+
+ l.keyListeners.Range(func(key, _ any) bool {
+ path := key.(string)
+ err := l.cache.ensureBusinessWatchWithRetry(path, func()
(watchRegistration, error) {
+ return l.registerWatcher(path)
+ }, 1)
+ if err != nil {
+ logger.Warnf("[ConfigCenter][Zookeeper] restore
configuration watcher failed, path=%s err=%v", path, err)
+ return true
+ }
+ pathLock := l.cache.pathLock(path)
+ pathLock.Lock()
+ if !l.hasListeners(path) {
+ l.cache.releaseBusinessWatchLocked(path)
+ }
+ pathLock.Unlock()
+ return true
+ })
+}
+
+// WatchStateChanged consumes the current watch state and reserves its renewal.
+// It returns whether the event loop should use ExistsW to register a new watch
+// or plain Exists when the path no longer has watch ownership.
+func (l *CacheListener) WatchStateChanged(path string) bool {
+ if l.cache == nil {
+ return true
+ }
+ pathLock := l.cache.pathLock(path)
+ pathLock.Lock()
+ defer pathLock.Unlock()
+ generation, watchState, registerWatch :=
l.cache.beginWatchRenewalLocked(path, l.hasListeners(path))
+ l.eventGeneration.Store(path, watchEventState{
+ generation: generation,
+ sessionID: watchState.sessionID,
+ auto: watchState.auto,
+ token: watchState.pendingOpToken(),
+ })
+ return registerWatch
+}
+
+// WatchRegistered validates a renewed watch against the event generation,
+// operation token, and ZooKeeper session. The result tells the event loop to
+// accept the read, reload without another watch, or discard the stale event.
+func (l *CacheListener) WatchRegistered(path string, events <-chan zk.Event,
beforeSessionID, afterSessionID int64) zookeeper.WatchRegistrationResult {
+ if l.cache == nil {
+ return zookeeper.WatchRegistrationAccepted
+ }
+ pathLock := l.cache.pathLock(path)
+ pathLock.Lock()
+ state, ok := l.eventGeneration.Load(path)
+ if !ok {
+ pathLock.Unlock()
+ return zookeeper.WatchRegistrationDiscarded
+ }
+ eventState := state.(watchEventState)
+ registration := watchRegistration{
+ events: events,
+ beforeSessionID: beforeSessionID,
+ afterSessionID: afterSessionID,
}
+ stored := l.cache.finishWatchRegistrationLocked(path,
eventState.generation, eventState.token, registration)
+ currentGeneration := l.cache.isCurrentGeneration(eventState.generation)
+ if stored && registration.sessionStable() && currentGeneration {
+ pathLock.Unlock()
+ return zookeeper.WatchRegistrationAccepted
+ }
+ reload := eventState.auto || (stored && !currentGeneration)
+ if reload {
+ l.eventGeneration.Store(path, eventState)
+ } else {
+ l.eventGeneration.Delete(path)
+ }
+ pathLock.Unlock()
+
+ if reload {
+ return zookeeper.WatchRegistrationReload
+ }
+ if l.retryBusinessWatch(path, eventState.sessionID,
eventState.generation) {
+ pathLock.Lock()
+ if _, ok := l.eventGeneration.Load(path); !ok {
+ l.eventGeneration.Store(path, eventState)
+ }
+ pathLock.Unlock()
+ return zookeeper.WatchRegistrationReload
+ }
+ return zookeeper.WatchRegistrationDiscarded
+}
+
+// WatchStateChangeFailed releases a pending renewal. If business listeners
+// remain and the connection has a live session, it may retry the watch after
+// the cache generation or ZooKeeper session has changed.
+func (l *CacheListener) WatchStateChangeFailed(path string) {
+ if l.cache == nil {
+ return
+ }
+ pathLock := l.cache.pathLock(path)
+ pathLock.Lock()
+ state, ok := l.eventGeneration.LoadAndDelete(path)
+ if !ok {
+ pathLock.Unlock()
+ return
+ }
+ l.cache.cancelPendingLocked(path)
+ pathLock.Unlock()
+ eventState := state.(watchEventState)
+ l.retryBusinessWatch(path, eventState.sessionID, eventState.generation)
+}
+
+func (l *CacheListener) retryBusinessWatch(path string, previousSessionID
int64, previousGeneration uint64) bool {
+ pathLock := l.cache.pathLock(path)
+ pathLock.Lock()
+ hasListeners := l.hasListeners(path)
+ pathLock.Unlock()
+ if !hasListeners || l.zkEventListener == nil ||
+ l.zkEventListener.Client == nil ||
l.zkEventListener.Client.Conn == nil {
+ return false
+ }
+
+ conn := l.zkEventListener.Client.Conn
+ currentGeneration, _ := l.cache.snapshot(path)
+ if conn.State() != zk.StateHasSession ||
+ (conn.SessionID() == previousSessionID && currentGeneration ==
previousGeneration) {
+ return false
+ }
+ if err := l.cache.ensureBusinessWatchWithRetry(path, func()
(watchRegistration, error) {
+ return l.registerWatcher(path)
+ }, 1); err != nil {
+ logger.Warnf("[ConfigCenter][Zookeeper] retry configuration
watcher failed, path=%s err=%v", path, err)
+ return false
+ }
+ pathLock = l.cache.pathLock(path)
+ pathLock.Lock()
+ defer pathLock.Unlock()
+ if !l.hasListeners(path) {
+ l.cache.releaseBusinessWatchLocked(path)
+ return false
+ }
+ return true
+}
+
+func (l *CacheListener) hasListeners(path string) bool {
+ _, ok := l.keyListeners.Load(path)
+ return ok
+}
+
+func (l *CacheListener) hasListener(path string, listener
config_center.ConfigurationListener) bool {
+ listeners, ok := l.keyListeners.Load(path)
+ if !ok {
+ return false
+ }
+ _, ok =
listeners.(map[config_center.ConfigurationListener]struct{})[listener]
+ return ok
}
// RemoveListener will delete a listener if loaded
func (l *CacheListener) RemoveListener(key string, listener
config_center.ConfigurationListener) {
+ if l.cache == nil {
+ l.removeListener(key, listener)
+ return
+ }
+
+ pathLock := l.cache.pathLock(key)
+ pathLock.Lock()
+ defer pathLock.Unlock()
+ removed, last := l.removeListener(key, listener)
+ if removed && last {
+ l.cache.releaseBusinessWatchLocked(key)
+ }
+}
+
+func (l *CacheListener) removeListener(key string, listener
config_center.ConfigurationListener) (bool, bool) {
listeners, loaded := l.keyListeners.Load(key)
- if loaded {
-
delete(listeners.(map[config_center.ConfigurationListener]struct{}), listener)
+ if !loaded {
+ return false, false
+ }
+ listenerSet :=
listeners.(map[config_center.ConfigurationListener]struct{})
+ if _, exists := listenerSet[listener]; !exists {
+ return false, false
+ }
+ delete(listenerSet, listener)
+ if len(listenerSet) != 0 {
+ l.keyListeners.Store(key, listenerSet)
+ return true, false
}
+ l.keyListeners.Delete(key)
+ return true, true
}
-// DataChange changes all listeners' event
+// DataChange updates the read-through cache before notifying a snapshot of the
+// business listeners. Events tied to an older generation cannot repopulate a
+// cache that has already been reset.
func (l *CacheListener) DataChange(event remoting.Event) bool {
- changeType := event.Action
- if event.Content == "" {
- changeType = remoting.EventTypeDel
+ if l.cache != nil {
+ entry := configCacheEntry{content: event.Content, exists: true}
+ if event.Action == remoting.EventTypeDel {
+ entry = configCacheEntry{exists: false}
+ }
+ pathLock := l.cache.pathLock(event.Path)
+ pathLock.Lock()
+ if state, ok := l.eventGeneration.LoadAndDelete(event.Path); ok
{
+ l.cache.storeAtGenerationLocked(event.Path,
state.(watchEventState).generation, entry)
+ } else {
+ l.cache.storeLocked(event.Path, entry)
+ }
+ pathLock.Unlock()
}
key, group := l.pathToKeyGroup(event.Path)
- defer metrics.Publish(metricsConfigCenter.NewIncMetricEvent(key, group,
changeType, metricsConfigCenter.Zookeeper))
- if listeners, ok := l.keyListeners.Load(event.Path); ok {
- for listener := range
listeners.(map[config_center.ConfigurationListener]struct{}) {
- listener.Process(&config_center.ConfigChangeEvent{
- Key: key,
- Value: event.Content,
- ConfigType: changeType,
- })
- }
- return true
+ defer metrics.Publish(metricsConfigCenter.NewIncMetricEvent(key, group,
event.Action, metricsConfigCenter.Zookeeper))
+ listeners := l.snapshotListeners(event.Path)
+ for _, listener := range listeners {
+ listener.Process(&config_center.ConfigChangeEvent{
+ Key: key,
+ Value: event.Content,
+ ConfigType: event.Action,
+ })
+ }
+ return len(listeners) != 0
+}
+
+func (l *CacheListener) snapshotListeners(path string)
[]config_center.ConfigurationListener {
+ if l.cache != nil {
+ pathLock := l.cache.pathLock(path)
+ pathLock.Lock()
+ defer pathLock.Unlock()
+ }
+
+ listeners, ok := l.keyListeners.Load(path)
+ if !ok {
+ return nil
+ }
+ listenerSet :=
listeners.(map[config_center.ConfigurationListener]struct{})
+ result := make([]config_center.ConfigurationListener, 0,
len(listenerSet))
+ for listener := range listenerSet {
+ result = append(result, listener)
}
- return false
+ return result
}
func (l *CacheListener) pathToKeyGroup(path string) (string, string) {
diff --git a/config_center/zookeeper/listener_test.go
b/config_center/zookeeper/listener_test.go
index 7835ad8b4..e84e1d4e1 100644
--- a/config_center/zookeeper/listener_test.go
+++ b/config_center/zookeeper/listener_test.go
@@ -18,12 +18,23 @@
package zookeeper
import (
+ "fmt"
+ "sync"
+ "sync/atomic"
"testing"
+ "time"
+)
+
+import (
+ "github.com/go-zookeeper/zk"
+
+ "github.com/stretchr/testify/require"
)
import (
"dubbo.apache.org/dubbo-go/v3/config_center"
"dubbo.apache.org/dubbo-go/v3/remoting"
+ remotingzookeeper "dubbo.apache.org/dubbo-go/v3/remoting/zookeeper"
)
type recListener struct {
@@ -50,7 +61,8 @@ func TestCacheListenerDataChange(t *testing.T) {
}
func TestCacheListenerDataChangeEmptyContent(t *testing.T) {
- l := &CacheListener{rootPath: "/dubbo/config"}
+ cache := newConfigCache(time.Minute)
+ l := &CacheListener{rootPath: "/dubbo/config", cache: &cache}
path := "/dubbo/config/group/app"
rec := &recListener{}
l.keyListeners.Store(path,
map[config_center.ConfigurationListener]struct{}{rec: {}})
@@ -59,9 +71,375 @@ func TestCacheListenerDataChangeEmptyContent(t *testing.T) {
if !ok {
t.Fatalf("expected listeners to be notified")
}
- if len(rec.events) != 1 || rec.events[0].ConfigType !=
remoting.EventTypeDel {
+ if len(rec.events) != 1 || rec.events[0].ConfigType !=
remoting.EventTypeAdd {
t.Fatalf("unexpected events %+v", rec.events)
}
+ entry, ok := cache.getFresh(path)
+ if !ok || !entry.exists || entry.content != "" {
+ t.Fatalf("empty configuration should be cached as existing:
%+v", entry)
+ }
+
+ l.DataChange(remoting.Event{Path: path, Action: remoting.EventTypeDel})
+ if len(rec.events) != 2 || rec.events[1].ConfigType !=
remoting.EventTypeDel {
+ t.Fatalf("unexpected events %+v", rec.events)
+ }
+ entry, ok = cache.getFresh(path)
+ if !ok || entry.exists {
+ t.Fatalf("deleted configuration should be cached as missing:
%+v", entry)
+ }
+}
+
+func TestCacheListenerIgnoresEventAcrossReset(t *testing.T) {
+ cache := newConfigCache(time.Minute)
+ l := &CacheListener{rootPath: "/dubbo/config", cache: &cache}
+ path := "/dubbo/config/group/app"
+ cache.setWatch(path, configWatchState{registered: true, auto: true,
sessionID: 1})
+ require.True(t, l.WatchStateChanged(path))
+ cache.reset(2)
+ require.Equal(t, remotingzookeeper.WatchRegistrationReload,
+ l.WatchRegistered(path, make(chan zk.Event, 1), 1, 1))
+
+ _, ok := cache.getFresh(path)
+ if ok {
+ t.Fatal("event started before reset should not repopulate
cache")
+ }
+ _, watchState := cache.snapshot(path)
+ if watchState.tracked() {
+ t.Fatal("event started before reset should not reactivate watch
state")
+ }
+}
+
+func TestCacheListenerReloadDataChangeRejectsResetGeneration(t *testing.T) {
+ cache := newConfigCache(time.Minute)
+ path := "/dubbo/config/group/app"
+ cache.reset(1)
+ cache.setWatch(path, configWatchState{registered: true, auto: true,
sessionID: 1})
+ l := &CacheListener{rootPath: "/dubbo/config", cache: &cache}
+
+ require.True(t, l.WatchStateChanged(path))
+ eventStateValue, ok := l.eventGeneration.Load(path)
+ require.True(t, ok)
+ require.Equal(t, uint64(1),
eventStateValue.(watchEventState).generation)
+
+ cache.reset(2)
+ events := make(chan zk.Event, 1)
+ events <- zk.Event{Type: zk.EventNotWatching}
+ close(events)
+ require.Equal(t, remotingzookeeper.WatchRegistrationReload,
+ l.WatchRegistered(path, events, 1, 1))
+ _, ok = l.eventGeneration.Load(path)
+ require.True(t, ok)
+
+ l.DataChange(remoting.Event{Path: path, Action:
remoting.EventTypeUpdate, Content: "stale"})
+ _, ok = cache.getFresh(path)
+ require.False(t, ok)
+ _, ok = l.eventGeneration.Load(path)
+ require.False(t, ok)
+}
+
+func TestAddListenerPromotesAutoWatch(t *testing.T) {
+ cache := newConfigCache(time.Minute)
+ path := "/dubbo/config/group/app"
+ cache.setWatch(path, configWatchState{registered: true, auto: true,
sessionID: 1})
+ l := &CacheListener{cache: &cache}
+ rec := &recListener{}
+
+ require.NotPanics(t, func() {
+ l.AddListener(path, rec)
+ })
+
+ _, watchState := cache.snapshot(path)
+ require.True(t, watchState.registered)
+ require.False(t, watchState.auto)
+ require.Zero(t, cache.autoWatchCount)
+ listeners, ok := l.keyListeners.Load(path)
+ require.True(t, ok)
+ _, ok =
listeners.(map[config_center.ConfigurationListener]struct{})[rec]
+ require.True(t, ok)
+}
+
+func TestAddListenerReactivatesRetiredWatch(t *testing.T) {
+ cache := newConfigCache(time.Minute)
+ path := "/dubbo/config/group/app"
+ cache.setWatch(path, configWatchState{
+ registered: true,
+ retired: true,
+ sessionID: 1,
+ })
+ l := &CacheListener{cache: &cache}
+ rec := &recListener{}
+
+ l.AddListener(path, rec)
+
+ _, watchState := cache.snapshot(path)
+ require.True(t, watchState.registered)
+ require.False(t, watchState.retired)
+ require.False(t, watchState.auto)
+ _, ok := l.keyListeners.Load(path)
+ require.True(t, ok)
+}
+
+func TestCacheListenerConcurrentAddListenerSharesPendingRegistration(t
*testing.T) {
+ cache := newConfigCache(time.Minute)
+ l := newCacheListener("/dubbo/config", nil, &cache)
+ path := "/dubbo/config/group/app"
+ first := &recListener{}
+ second := &recListener{}
+ registerStarted := make(chan struct{})
+ releaseRegister := make(chan struct{})
+ var registrations atomic.Int32
+ register := func() (watchRegistration, error) {
+ if registrations.Add(1) == 1 {
+ close(registerStarted)
+ <-releaseRegister
+ }
+ return newTestWatchRegistration(1), nil
+ }
+
+ var wg sync.WaitGroup
+ wg.Go(func() { l.addListenerWithRegister(path, first, register) })
+ <-registerStarted
+ wg.Go(func() { l.addListenerWithRegister(path, second, register) })
+ time.Sleep(10 * time.Millisecond)
+ require.Equal(t, int32(1), registrations.Load())
+
+ close(releaseRegister)
+ wg.Wait()
+ require.Equal(t, int32(1), registrations.Load())
+ listeners, ok := l.keyListeners.Load(path)
+ require.True(t, ok)
+ listenerSet :=
listeners.(map[config_center.ConfigurationListener]struct{})
+ require.Len(t, listenerSet, 2)
+ _, watchState := cache.snapshot(path)
+ require.True(t, watchState.registered)
+ require.False(t, watchState.auto)
+}
+
+func TestAddListenerRegistrationFailureDoesNotRemoveExistingListener(t
*testing.T) {
+ cache := newConfigCache(time.Minute)
+ l := newCacheListener("/dubbo/config", nil, &cache)
+ path := "/dubbo/config/group/app"
+ rec := &recListener{}
+ l.storeListener(path, rec)
+
+ l.addListenerWithRegister(path, rec, func() (watchRegistration, error) {
+ return watchRegistration{}, errWatchRegistrationStale
+ })
+
+ listeners, ok := l.keyListeners.Load(path)
+ require.True(t, ok)
+ listenerSet :=
listeners.(map[config_center.ConfigurationListener]struct{})
+ _, ok = listenerSet[rec]
+ require.True(t, ok)
+}
+
+func TestAddListenerStopsRetryWhenListenerIsRemoved(t *testing.T) {
+ cache := newConfigCache(time.Minute)
+ l := newCacheListener("/dubbo/config", nil, &cache)
+ path := "/dubbo/config/group/app"
+ rec := &recListener{}
+ registerStarted := make(chan struct{})
+ releaseRegister := make(chan struct{})
+ var registrations atomic.Int32
+ register := func() (watchRegistration, error) {
+ registrations.Add(1)
+ close(registerStarted)
+ <-releaseRegister
+ events := make(chan zk.Event, 1)
+ events <- zk.Event{Type: zk.EventNotWatching}
+ close(events)
+ return watchRegistration{
+ events: events,
+ beforeSessionID: 1,
+ afterSessionID: 2,
+ }, nil
+ }
+
+ done := make(chan struct{})
+ go func() {
+ l.addListenerWithRegister(path, rec, register)
+ close(done)
+ }()
+ <-registerStarted
+ l.RemoveListener(path, rec)
+ close(releaseRegister)
+ select {
+ case <-done:
+ case <-time.After(time.Second):
+ t.Fatal("AddListener did not stop after listener removal")
+ }
+
+ require.Equal(t, int32(1), registrations.Load())
+ _, ok := l.keyListeners.Load(path)
+ require.False(t, ok)
+ _, watchState := cache.snapshot(path)
+ require.False(t, watchState.tracked())
+}
+
+func TestRemovingOneListenerDoesNotReleaseSharedBusinessWatch(t *testing.T) {
+ cache := newConfigCache(time.Minute)
+ l := newCacheListener("/dubbo/config", nil, &cache)
+ path := "/dubbo/config/group/app"
+ first := &recListener{}
+ second := &recListener{}
+ registerStarted := make(chan struct{})
+ releaseRegister := make(chan struct{})
+ var registrations atomic.Int32
+ register := func() (watchRegistration, error) {
+ if registrations.Add(1) == 1 {
+ close(registerStarted)
+ <-releaseRegister
+ }
+ return newTestWatchRegistration(1), nil
+ }
+
+ var wg sync.WaitGroup
+ wg.Go(func() { l.addListenerWithRegister(path, first, register) })
+ <-registerStarted
+ wg.Go(func() { l.addListenerWithRegister(path, second, register) })
+ time.Sleep(10 * time.Millisecond)
+ l.RemoveListener(path, first)
+ close(releaseRegister)
+ wg.Wait()
+
+ require.Equal(t, int32(1), registrations.Load())
+ require.True(t, l.hasListener(path, second))
+ _, watchState := cache.snapshot(path)
+ require.True(t, watchState.registered)
+ require.False(t, watchState.auto)
+}
+
+func TestWatchStateChangedUsesCurrentBusinessOwnership(t *testing.T) {
+ cache := newConfigCache(time.Minute)
+ path := "/dubbo/config/group/app"
+ require.True(t, cache.setWatch(path, configWatchState{
+ registered: true,
+ auto: true,
+ sessionID: 1,
+ }))
+ l := &CacheListener{cache: &cache}
+ rec := &recListener{}
+
+ l.AddListener(path, rec)
+ require.True(t, l.WatchStateChanged(path))
+
+ _, watchState := cache.snapshot(path)
+ require.True(t, watchState.pending)
+ require.False(t, watchState.auto)
+ require.Zero(t, cache.autoWatchReservations)
+}
+
+func TestAddListenerRegistersBusinessWatchAtAutoWatchLimit(t *testing.T) {
+ client, _ := newZookeeperTestClient(t, "business-watch-limit")
+ root := newTestRoot(t, client)
+
+ cache := newConfigCache(time.Minute)
+ for i := range maxAutoWatches {
+ require.True(t, cache.setWatch(fmt.Sprintf("/auto/%d", i),
configWatchState{
+ registered: true,
+ auto: true,
+ sessionID: 1,
+ }))
+ }
+ zkListener := remotingzookeeper.NewZkEventListener(client)
+ defer zkListener.Close()
+ l := newCacheListener(root, zkListener, &cache)
+ path := root + "/group/app"
+ rec := &recListener{}
+
+ l.AddListener(path, rec)
+
+ _, watchState := cache.snapshot(path)
+ require.True(t, watchState.registered)
+ require.False(t, watchState.auto)
+ require.Equal(t, maxAutoWatches, cache.autoWatchCount)
+ require.Zero(t, cache.autoWatchReservations)
+ listeners, ok := l.keyListeners.Load(path)
+ require.True(t, ok)
+ _, ok =
listeners.(map[config_center.ConfigurationListener]struct{})[rec]
+ require.True(t, ok)
+}
+
+func TestCacheListenerPreservesAutoWatchOwnershipOnRenewal(t *testing.T) {
+ cache := newConfigCache(time.Minute)
+ path := "/dubbo/config/group/app"
+ cache.setWatch(path, configWatchState{registered: true, auto: true,
sessionID: 1})
+ l := &CacheListener{cache: &cache}
+
+ require.True(t, l.WatchStateChanged(path))
+ _, watchState := cache.snapshot(path)
+ require.False(t, watchState.registered)
+ require.True(t, watchState.pending)
+ require.True(t, watchState.auto)
+ require.Zero(t, cache.autoWatchCount)
+ require.Equal(t, 1, cache.autoWatchReservations)
+
+ require.Equal(t, remotingzookeeper.WatchRegistrationAccepted,
+ l.WatchRegistered(path, make(chan zk.Event, 1), 1, 1))
+ _, watchState = cache.snapshot(path)
+ require.True(t, watchState.registered)
+ require.True(t, watchState.auto)
+ require.False(t, watchState.pending)
+ require.Equal(t, 1, cache.autoWatchCount)
+ require.Zero(t, cache.autoWatchReservations)
+}
+
+func TestCacheListenerRetriesInvalidatedWatchInNewSession(t *testing.T) {
+ client, _ := newZookeeperTestClient(t,
"retry-invalidated-business-watch")
+ root := newTestRoot(t, client)
+ zkListener := remotingzookeeper.NewZkEventListener(client)
+ defer zkListener.Close()
+
+ cache := newConfigCache(time.Minute)
+ currentSessionID := client.Conn.SessionID()
+ cache.reset(currentSessionID)
+ path := root + "/group/app"
+ previousSessionID := currentSessionID - 1
+ require.True(t, cache.setWatch(path, configWatchState{
+ pending: true,
+ sessionID: previousSessionID,
+ }))
+ l := newCacheListener(root, zkListener, &cache)
+ l.keyListeners.Store(path,
map[config_center.ConfigurationListener]struct{}{&recListener{}: {}})
+ l.eventGeneration.Store(path, watchEventState{
+ generation: cache.generation,
+ sessionID: previousSessionID,
+ })
+ events := make(chan zk.Event, 1)
+ events <- zk.Event{Type: zk.EventNotWatching}
+ close(events)
+
+ require.Equal(t, remotingzookeeper.WatchRegistrationReload,
+ l.WatchRegistered(path, events, previousSessionID,
currentSessionID))
+ _, watchState := cache.snapshot(path)
+ require.True(t, watchState.registered)
+ require.False(t, watchState.pending)
+ require.False(t, watchState.auto)
+ require.Equal(t, currentSessionID, watchState.sessionID)
+}
+
+func TestCacheListenerDiscardsStaleBusinessRegistrationWithoutListener(t
*testing.T) {
+ cache := newConfigCache(time.Minute)
+ path := "/dubbo/config/group/app"
+ require.True(t, cache.setWatch(path, configWatchState{
+ pending: true,
+ sessionID: 1,
+ }))
+ l := &CacheListener{cache: &cache}
+ l.eventGeneration.Store(path, watchEventState{
+ generation: cache.generation,
+ sessionID: 1,
+ })
+ events := make(chan zk.Event, 1)
+ events <- zk.Event{Type: zk.EventNotWatching}
+ close(events)
+
+ result := l.WatchRegistered(path, events, 1, 2)
+ require.Equal(t, remotingzookeeper.WatchRegistrationDiscarded, result)
+ _, watchState := cache.snapshot(path)
+ require.False(t, watchState.tracked())
+ _, ok := l.eventGeneration.Load(path)
+ require.False(t, ok)
}
func TestCacheListenerPathToKeyGroup(t *testing.T) {
@@ -73,14 +451,74 @@ func TestCacheListenerPathToKeyGroup(t *testing.T) {
}
func TestCacheListenerRemoveListener(t *testing.T) {
- l := &CacheListener{}
+ cache := newConfigCache(time.Minute)
+ l := &CacheListener{cache: &cache}
key := "k"
rec := &recListener{}
+ cache.setWatch(key, configWatchState{registered: true, sessionID: 1})
l.keyListeners.Store(key,
map[config_center.ConfigurationListener]struct{}{rec: {}})
+
l.RemoveListener(key, rec)
- if m, ok := l.keyListeners.Load(key); ok {
- if _, exists :=
m.(map[config_center.ConfigurationListener]struct{})[rec]; exists {
- t.Fatalf("listener should be removed")
- }
+
+ _, ok := l.keyListeners.Load(key)
+ require.False(t, ok)
+ _, watchState := cache.snapshot(key)
+ require.True(t, watchState.registered)
+ require.True(t, watchState.auto)
+ require.Equal(t, 1, cache.autoWatchCount)
+}
+
+func TestCacheListenerRemoveListenerRetiresWatchAtAutoLimit(t *testing.T) {
+ cache := newConfigCache(time.Minute)
+ for i := range maxAutoWatches {
+ require.True(t, cache.setWatch(fmt.Sprintf("/auto/%d", i),
configWatchState{
+ registered: true,
+ auto: true,
+ sessionID: 1,
+ }))
}
+ path := "/dubbo/config/group/app"
+ require.True(t, cache.setWatch(path, configWatchState{
+ registered: true,
+ sessionID: 1,
+ }))
+ l := newCacheListener("/dubbo/config", nil, &cache)
+ rec := &recListener{}
+ l.keyListeners.Store(path,
map[config_center.ConfigurationListener]struct{}{rec: {}})
+
+ l.RemoveListener(path, rec)
+
+ _, ok := l.keyListeners.Load(path)
+ require.False(t, ok)
+ _, watchState := cache.snapshot(path)
+ require.True(t, watchState.registered)
+ require.True(t, watchState.retired)
+ require.False(t, watchState.auto)
+ require.Equal(t, maxAutoWatches, cache.autoWatchCount)
+
+ require.False(t, l.WatchStateChanged(path))
+ _, watchState = cache.snapshot(path)
+ require.False(t, watchState.tracked())
+ require.False(t, l.DataChange(remoting.Event{Path: path, Action:
remoting.EventTypeUpdate, Content: "value"}))
+ entry, ok := cache.getFresh(path)
+ require.True(t, ok)
+ require.Equal(t, "value", entry.content)
+}
+
+func TestCacheListenerResidualEventDoesNotRenewWatch(t *testing.T) {
+ cache := newConfigCache(time.Minute)
+ path := "/dubbo/config/group/app"
+ l := &CacheListener{rootPath: "/dubbo/config", cache: &cache}
+
+ require.False(t, l.WatchStateChanged(path))
+ _, watchState := cache.snapshot(path)
+ require.False(t, watchState.tracked())
+ require.False(t, l.DataChange(remoting.Event{Path: path, Action:
remoting.EventTypeUpdate, Content: "value"}))
+
+ entry, ok := cache.getFresh(path)
+ require.True(t, ok)
+ require.True(t, entry.exists)
+ require.Equal(t, "value", entry.content)
+ _, ok = l.eventGeneration.Load(path)
+ require.False(t, ok)
}
diff --git a/remoting/zookeeper/listener.go b/remoting/zookeeper/listener.go
index 782b84797..c50a91951 100644
--- a/remoting/zookeeper/listener.go
+++ b/remoting/zookeeper/listener.go
@@ -56,6 +56,27 @@ type ZkEventListener struct {
exit chan struct{}
}
+// WatchRegistrationResult tells the configuration event loop how to continue
+// after a listener validates a newly registered ZooKeeper watch.
+type WatchRegistrationResult uint8
+
+const (
+ // WatchRegistrationAccepted keeps the watch and processes the current
read.
+ WatchRegistrationAccepted WatchRegistrationResult = iota
+ // WatchRegistrationReload re-reads the node without registering
another watch.
+ WatchRegistrationReload
+ // WatchRegistrationDiscarded stops processing a stale event.
+ WatchRegistrationDiscarded
+)
+
+// configurationWatchStateListener lets the config center validate watch state
+// while the remoting layer remains responsible for ZooKeeper reads and events.
+type configurationWatchStateListener interface {
+ WatchStateChanged(path string) bool
+ WatchRegistered(path string, events <-chan zk.Event, beforeSessionID,
afterSessionID int64) WatchRegistrationResult
+ WatchStateChangeFailed(path string)
+}
+
// NewZkEventListener returns a EventListener instance
func NewZkEventListener(client *gxzookeeper.ZookeeperClient) *ZkEventListener {
return &ZkEventListener{
@@ -84,45 +105,15 @@ func (l *ZkEventListener) ListenServiceNodeEvent(zkPath
string, listener remotin
func (l *ZkEventListener) ListenConfigurationEvent(zkPath string, listener
remoting.DataListener) {
l.wg.Add(1)
go func(zkPath string, listener remoting.DataListener) {
+ defer l.wg.Done()
var eventChan = make(chan zk.Event, 16)
l.Client.RegisterEvent(zkPath, eventChan)
+ watchStateListener, tracksWatchState :=
listener.(configurationWatchStateListener)
for {
select {
case event := <-eventChan:
logger.Infof("[Remoting][Zookeeper]Receive
configuration change event:%#v", event)
- if event.Type == zk.EventNodeChildrenChanged ||
event.Type == zk.EventNotWatching {
- continue
- }
- // 1. Re-set watcher for the zk node
- _, _, _, err :=
l.Client.Conn.ExistsW(event.Path)
- if err != nil {
-
logger.Warnf("[Remoting][Zookeeper]Re-set watcher error, err=%v", err)
- continue
- }
-
- action := remoting.EventTypeAdd
- var content string
- if event.Type == zk.EventNodeDeleted {
- action = remoting.EventTypeDel
- } else {
- // 2. Try to get new configuration
value of the zk node
- // Notice: The order of step 1 and step
2 cannot be swapped, if you get value(with timestamp t1)
- // before re-set the watcher(with
timestamp t2), and some one change the data of the zk node after
- // t2 but before t1, you may get the
old value, and the new value will not trigger the event.
- contentBytes, _, err :=
l.Client.Conn.Get(event.Path)
- if err != nil {
-
logger.Warnf("[Remoting][Zookeeper] get config value error, err=%v", err)
- continue
- }
- content = string(contentBytes)
- logger.Debugf("[Remoting][Zookeeper]
successfully get new config value=%s", string(content))
- }
-
- listener.DataChange(remoting.Event{
- Path: event.Path,
- Action: remoting.EventType(action),
- Content: content,
- })
+ l.processConfigurationEvent(event, listener,
watchStateListener, tracksWatchState)
case <-l.exit:
return
}
@@ -131,6 +122,135 @@ func (l *ZkEventListener) ListenConfigurationEvent(zkPath
string, listener remot
}(zkPath, listener)
}
+type configurationEventReader struct {
+ exists func(registerWatch bool) (bool, <-chan zk.Event, int64, int64,
error)
+ content func() ([]byte, int64, int64, error)
+}
+
+// processConfigurationEvent renews a one-shot watch when requested before
+// reading the latest value, then lets watch-aware listeners reject results
from
+// stale sessions.
+func (l *ZkEventListener) processConfigurationEvent(
+ event zk.Event,
+ listener remoting.DataListener,
+ watchStateListener configurationWatchStateListener,
+ tracksWatchState bool,
+) {
+ reader := configurationEventReader{
+ exists: func(registerWatch bool) (bool, <-chan zk.Event, int64,
int64, error) {
+ beforeSessionID := l.Client.Conn.SessionID()
+ var (
+ exists bool
+ watchEvents <-chan zk.Event
+ err error
+ )
+ if registerWatch {
+ exists, _, watchEvents, err =
l.Client.Conn.ExistsW(event.Path)
+ } else {
+ exists, _, err =
l.Client.Conn.Exists(event.Path)
+ }
+ return exists, watchEvents, beforeSessionID,
l.Client.Conn.SessionID(), err
+ },
+ content: func() ([]byte, int64, int64, error) {
+ beforeSessionID := l.Client.Conn.SessionID()
+ content, _, err := l.Client.Conn.Get(event.Path)
+ return content, beforeSessionID,
l.Client.Conn.SessionID(), err
+ },
+ }
+ processConfigurationEvent(event, listener, watchStateListener,
tracksWatchState, reader)
+}
+
+func processConfigurationEvent(
+ event zk.Event,
+ listener remoting.DataListener,
+ watchStateListener configurationWatchStateListener,
+ tracksWatchState bool,
+ reader configurationEventReader,
+) {
+ if event.Type == zk.EventNotWatching || event.Type ==
zk.EventNodeChildrenChanged {
+ return
+ }
+ registerWatch := true
+ if tracksWatchState {
+ registerWatch = watchStateListener.WatchStateChanged(event.Path)
+ }
+
+ // Re-set the watcher before reading the value so a concurrent update
cannot
+ // occur between the read and watch registration.
+ exists, watchEvents, beforeSessionID, afterSessionID, err :=
reader.exists(registerWatch)
+ if err != nil {
+ if tracksWatchState {
+ watchStateListener.WatchStateChangeFailed(event.Path)
+ }
+ logger.Warnf("[Remoting][Zookeeper]Re-set watcher error,
err=%v", err)
+ return
+ }
+ if tracksWatchState && registerWatch {
+ registrationResult := watchStateListener.WatchRegistered(
+ event.Path, watchEvents, beforeSessionID,
afterSessionID,
+ )
+ if registrationResult == WatchRegistrationDiscarded {
+ return
+ }
+ if registrationResult == WatchRegistrationAccepted &&
+ !sessionStable(beforeSessionID, afterSessionID) {
+ watchStateListener.WatchStateChangeFailed(event.Path)
+ return
+ }
+ if registrationResult == WatchRegistrationReload {
+ exists, _, beforeSessionID, afterSessionID, err =
reader.exists(false)
+ if err != nil {
+
watchStateListener.WatchStateChangeFailed(event.Path)
+ logger.Warnf("[Remoting][Zookeeper] reload
config existence error, err=%v", err)
+ return
+ }
+ if !sessionStable(beforeSessionID, afterSessionID) {
+
watchStateListener.WatchStateChangeFailed(event.Path)
+ return
+ }
+ }
+ }
+ if !tracksWatchState || !registerWatch {
+ if !sessionStable(beforeSessionID, afterSessionID) {
+ if tracksWatchState {
+
watchStateListener.WatchStateChangeFailed(event.Path)
+ }
+ return
+ }
+ }
+
+ action := remoting.EventTypeDel
+ var content string
+ if exists {
+ action = remoting.EventTypeAdd
+ contentBytes, beforeSessionID, afterSessionID, err :=
reader.content()
+ if err != nil {
+ if tracksWatchState {
+
watchStateListener.WatchStateChangeFailed(event.Path)
+ }
+ logger.Warnf("[Remoting][Zookeeper] get config value
error, err=%v", err)
+ return
+ }
+ if !sessionStable(beforeSessionID, afterSessionID) {
+ if tracksWatchState {
+
watchStateListener.WatchStateChangeFailed(event.Path)
+ }
+ return
+ }
+ content = string(contentBytes)
+ logger.Debugf("[Remoting][Zookeeper] successfully get new
config value=%s", content)
+ }
+
+ listener.DataChange(remoting.Event{Path: event.Path, Action: action,
Content: content})
+}
+
+// sessionStable reports whether a ZooKeeper operation completed in the session
+// in which it started. A zero pair represents a reader without session data.
+func sessionStable(beforeSessionID, afterSessionID int64) bool {
+ return (beforeSessionID == 0 && afterSessionID == 0) ||
+ (beforeSessionID != 0 && beforeSessionID == afterSessionID)
+}
+
// listenServiceNodeEvent watches a single zk node and reports changes via
listener.
func (l *ZkEventListener) listenServiceNodeEvent(zkPath string, listener
...remoting.DataListener) bool {
l.pathMapLock.Lock()
diff --git a/remoting/zookeeper/listener_test.go
b/remoting/zookeeper/listener_test.go
index 64e042b85..d4f6f5844 100644
--- a/remoting/zookeeper/listener_test.go
+++ b/remoting/zookeeper/listener_test.go
@@ -20,14 +20,221 @@ package zookeeper
import (
"net/url"
"testing"
+ "time"
)
import (
+ gxzookeeper "github.com/dubbogo/gost/database/kv/zk"
+
+ "github.com/go-zookeeper/zk"
+
"github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+import (
+ "dubbo.apache.org/dubbo-go/v3/remoting"
)
+type recordingDataListener struct {
+ events []remoting.Event
+}
+
+func (l *recordingDataListener) DataChange(event remoting.Event) bool {
+ l.events = append(l.events, event)
+ return true
+}
+
+type recordingWatchStateListener struct {
+ registerWatch bool
+ result WatchRegistrationResult
+ changed int
+ registered int
+ failed int
+}
+
+func (l *recordingWatchStateListener) WatchStateChanged(string) bool {
+ l.changed++
+ return l.registerWatch
+}
+
+func (l *recordingWatchStateListener) WatchRegistered(string, <-chan zk.Event,
int64, int64) WatchRegistrationResult {
+ l.registered++
+ return l.result
+}
+
+func (l *recordingWatchStateListener) WatchStateChangeFailed(string) {
+ l.failed++
+}
+
func TestZkPath(t *testing.T) {
zkPath := "io.grpc.examples.helloworld.GreeterGrpc$IGreeter"
zkPath = url.QueryEscape(zkPath)
assert.Equal(t, "io.grpc.examples.helloworld.GreeterGrpc%24IGreeter",
zkPath)
}
+
+func TestListenConfigurationEventStopsOnClose(t *testing.T) {
+ client, err := gxzookeeper.NewZookeeperClient(
+ "remoting-listener-event-loop",
+ []string{"127.0.0.1:0"},
+ false,
+ gxzookeeper.WithZkTimeOut(100*time.Millisecond),
+ )
+ require.NoError(t, err)
+ defer client.Close()
+
+ eventListener := NewZkEventListener(client)
+ eventListener.ListenConfigurationEvent("/config",
&recordingDataListener{})
+ // ListenConfigurationEvent registers asynchronously before waiting on
exit.
+ time.Sleep(50 * time.Millisecond)
+
+ closed := make(chan struct{})
+ go func() {
+ eventListener.Close()
+ close(closed)
+ }()
+ select {
+ case <-closed:
+ case <-time.After(2 * time.Second):
+ t.Fatal("ListenConfigurationEvent goroutine did not exit after
Close")
+ }
+}
+
+func TestProcessConfigurationEventRejectsUnstableMissingNodeRead(t *testing.T)
{
+ dataListener := &recordingDataListener{}
+ watchListener := &recordingWatchStateListener{}
+ reader := configurationEventReader{
+ exists: func(bool) (bool, <-chan zk.Event, int64, int64, error)
{
+ return false, nil, 1, 2, nil
+ },
+ content: func() ([]byte, int64, int64, error) {
+ t.Fatalf("content read should not happen for a missing
node")
+ return nil, 0, 0, nil
+ },
+ }
+
+ processConfigurationEvent(
+ zk.Event{Type: zk.EventNodeDataChanged, Path: "/config"},
+ dataListener,
+ watchListener,
+ true,
+ reader,
+ )
+
+ require.Empty(t, dataListener.events)
+ require.Equal(t, 1, watchListener.failed)
+}
+
+func TestProcessConfigurationEventHandlesMissingNodeAndWatchResults(t
*testing.T) {
+ t.Run("ttl fallback reports stable deletion", func(t *testing.T) {
+ dataListener := &recordingDataListener{}
+ watchListener := &recordingWatchStateListener{}
+ processConfigurationEvent(
+ zk.Event{Type: zk.EventNodeDataChanged, Path:
"/config"},
+ dataListener,
+ watchListener,
+ true,
+ configurationEventReader{
+ exists: func(register bool) (bool, <-chan
zk.Event, int64, int64, error) {
+ require.False(t, register)
+ return false, nil, 1, 1, nil
+ },
+ content: func() ([]byte, int64, int64, error) {
+ t.Fatalf("content read should not
happen for a missing node")
+ return nil, 0, 0, nil
+ },
+ },
+ )
+
+ require.Len(t, dataListener.events, 1)
+ require.Equal(t, remoting.EventTypeDel,
dataListener.events[0].Action)
+ require.Zero(t, watchListener.failed)
+ })
+
+ t.Run("accepted watch reports updated content", func(t *testing.T) {
+ dataListener := &recordingDataListener{}
+ watchListener := &recordingWatchStateListener{
+ registerWatch: true,
+ result: WatchRegistrationAccepted,
+ }
+ processConfigurationEvent(
+ zk.Event{Type: zk.EventNodeDataChanged, Path:
"/config"},
+ dataListener,
+ watchListener,
+ true,
+ configurationEventReader{
+ exists: func(register bool) (bool, <-chan
zk.Event, int64, int64, error) {
+ require.True(t, register)
+ return true, make(chan zk.Event), 1, 1,
nil
+ },
+ content: func() ([]byte, int64, int64, error) {
+ return []byte("value"), 1, 1, nil
+ },
+ },
+ )
+
+ require.Len(t, dataListener.events, 1)
+ require.Equal(t, remoting.EventTypeAdd,
dataListener.events[0].Action)
+ require.Equal(t, "value", dataListener.events[0].Content)
+ require.Equal(t, 1, watchListener.registered)
+ })
+
+ t.Run("discarded watch does not notify", func(t *testing.T) {
+ dataListener := &recordingDataListener{}
+ watchListener := &recordingWatchStateListener{
+ registerWatch: true,
+ result: WatchRegistrationDiscarded,
+ }
+ processConfigurationEvent(
+ zk.Event{Type: zk.EventNodeDataChanged, Path:
"/config"},
+ dataListener,
+ watchListener,
+ true,
+ configurationEventReader{
+ exists: func(bool) (bool, <-chan zk.Event,
int64, int64, error) {
+ return true, make(chan zk.Event), 1, 1,
nil
+ },
+ content: func() ([]byte, int64, int64, error) {
+ t.Fatalf("content read should not
happen for a discarded watch")
+ return nil, 0, 0, nil
+ },
+ },
+ )
+
+ require.Empty(t, dataListener.events)
+ })
+
+ t.Run("reload uses an ordinary stable existence read", func(t
*testing.T) {
+ dataListener := &recordingDataListener{}
+ watchListener := &recordingWatchStateListener{
+ registerWatch: true,
+ result: WatchRegistrationReload,
+ }
+ var calls int
+ processConfigurationEvent(
+ zk.Event{Type: zk.EventNodeDeleted, Path: "/config"},
+ dataListener,
+ watchListener,
+ true,
+ configurationEventReader{
+ exists: func(register bool) (bool, <-chan
zk.Event, int64, int64, error) {
+ calls++
+ if calls == 1 {
+ require.True(t, register)
+ return true, make(chan
zk.Event), 1, 1, nil
+ }
+ require.False(t, register)
+ return false, nil, 1, 1, nil
+ },
+ content: func() ([]byte, int64, int64, error) {
+ t.Fatalf("content read should not
happen after reload reports missing")
+ return nil, 0, 0, nil
+ },
+ },
+ )
+
+ require.Len(t, dataListener.events, 1)
+ require.Equal(t, remoting.EventTypeDel,
dataListener.events[0].Action)
+ require.Equal(t, 2, calls)
+ })
+}
