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 e506f0d4a fix(config_center/nacos): remove unused context.WithCancel
and call CancelListenConfig on last listener removal (#3441)
e506f0d4a is described below
commit e506f0d4a9401d346404ac697ff8eb4b53ac9f8e
Author: aias00 <[email protected]>
AuthorDate: Tue Aug 11 13:56:58 2026 +0800
fix(config_center/nacos): remove unused context.WithCancel and call
CancelListenConfig on last listener removal (#3441)
* fix(config_center/nacos): remove unused context.WithCancel and call
CancelListenConfig on last listener removal
The addListener() function created context.WithCancel but the cancel
function was never called, causing a goroutine and context leak.
Additionally, removeListener() never called CancelListenConfig to
unsubscribe from the nacos server, leaving stale config change
listeners active indefinitely.
Changes:
- Remove unused context.WithCancel calls in addListener()
- Replace context.CancelFunc value type with struct{} in keyListeners
map (the cancel function was never used)
- In removeListener(), call CancelListenConfig when the last listener
for a key is removed (matching Apollo and File implementations)
- Clean up keyListeners entry when no listeners remain
- Add nil client guard for test compatibility
Co-Authored-By: Claude <[email protected]>
* fix(config_center/nacos): use keyListenerSet with mutex for thread-safe
listener management
Address Copilot review feedback:
1. Race condition: Replace bare sync.Map with keyListenerSet struct that
uses sync.Mutex to make add/remove/check-empty atomic, preventing
concurrent addListener from racing with removeListener's emptiness
check and CancelListenConfig call.
2. DataId/Group consistency: Store the resolved group string in
keyListenerSet so that CancelListenConfig uses the exact same group
that was passed to ListenConfig, avoiding mismatch from re-derivation.
3. Delete keyListeners entry before CancelListenConfig to prevent new
addListener from finding a stale set after cancellation.
4. Use snapshot() for callback iteration to safely iterate while
mutations may occur.
Also add test for multi-listener removal behavior.
* fix(config_center/nacos): guard listener lifecycle race
* style(config_center/nacos): apply imports-formatter to concurrency test
Resolve the `make check-fmt` failure flagged in PR review: the new
listener_concurrency_test.go did not follow the repo's import grouping
order. Re-run `make fmt`; `check-fmt` now passes.
Co-Authored-By: Claude <[email protected]>
---------
Co-authored-by: Claude <[email protected]>
---
config_center/nacos/impl.go | 3 +-
config_center/nacos/listener.go | 111 ++++++--
config_center/nacos/listener_concurrency_test.go | 320 +++++++++++++++++++++++
config_center/nacos/listener_test.go | 42 ++-
4 files changed, 447 insertions(+), 29 deletions(-)
diff --git a/config_center/nacos/impl.go b/config_center/nacos/impl.go
index db672d796..003237ecb 100644
--- a/config_center/nacos/impl.go
+++ b/config_center/nacos/impl.go
@@ -57,7 +57,8 @@ type nacosDynamicConfiguration struct {
cltLock sync.Mutex
done chan struct{}
client *nacosClient.NacosConfigClient
- keyListeners sync.Map //
sync.Map[listenKey]*sync.Map[config_center.ConfigurationListener]context.CancelFunc
+ keyListeners sync.Map // sync.Map[listenKey]*keyListenerSet
+ listenerLock sync.Mutex
parser parser.ConfigurationParser
}
diff --git a/config_center/nacos/listener.go b/config_center/nacos/listener.go
index 542da3a17..c1b44ff05 100644
--- a/config_center/nacos/listener.go
+++ b/config_center/nacos/listener.go
@@ -18,7 +18,6 @@
package nacos
import (
- "context"
"sync"
)
@@ -37,29 +36,81 @@ import (
"dubbo.apache.org/dubbo-go/v3/remoting"
)
-func callback(listenersMap *sync.Map, _, group, dataId, data string) {
- listenersMap.Range(func(key, value any) bool {
-
key.(config_center.ConfigurationListener).Process(&config_center.ConfigChangeEvent{Key:
dataId, Value: data, ConfigType: remoting.EventTypeUpdate})
+// keyListenerSet holds the listeners for a single config key.
+// The mutex protects the listener map while callbacks take snapshots and
+// add/remove paths mutate it.
+type keyListenerSet struct {
+ mu sync.Mutex
+ listeners map[config_center.ConfigurationListener]struct{}
+ group string // resolved group used to register with nacos, stored
for consistent cancel
+}
+
+func newKeyListenerSet(group string) *keyListenerSet {
+ return &keyListenerSet{
+ listeners:
make(map[config_center.ConfigurationListener]struct{}),
+ group: group,
+ }
+}
+
+func (s *keyListenerSet) add(listener config_center.ConfigurationListener) {
+ s.mu.Lock()
+ s.listeners[listener] = struct{}{}
+ s.mu.Unlock()
+}
+
+// remove removes a listener and reports whether the set is now empty.
+// Callers that use the empty result to cancel the Nacos subscription must hold
+// nacosDynamicConfiguration.listenerLock.
+func (s *keyListenerSet) remove(listener config_center.ConfigurationListener)
bool {
+ s.mu.Lock()
+ delete(s.listeners, listener)
+ empty := len(s.listeners) == 0
+ s.mu.Unlock()
+ return empty
+}
+
+// snapshot returns a snapshot of the current listeners for safe iteration.
+func (s *keyListenerSet) snapshot() []config_center.ConfigurationListener {
+ s.mu.Lock()
+ snapshot := make([]config_center.ConfigurationListener, 0,
len(s.listeners))
+ for l := range s.listeners {
+ snapshot = append(snapshot, l)
+ }
+ s.mu.Unlock()
+ return snapshot
+}
+
+func callback(set *keyListenerSet, _, group, dataId, data string) {
+ for _, l := range set.snapshot() {
+ l.Process(&config_center.ConfigChangeEvent{Key: dataId, Value:
data, ConfigType: remoting.EventTypeUpdate})
metrics.Publish(metricsConfigCenter.NewIncMetricEvent(dataId,
group, remoting.EventTypeUpdate, metricsConfigCenter.Nacos))
- return true
- })
+ }
}
func (n *nacosDynamicConfiguration) addListener(key string, listener
config_center.ConfigurationListener) {
- rawListenersMap, loaded := n.keyListeners.Load(key)
+ group := n.resolvedGroup(n.url.GetParam(constant.NacosGroupKey,
constant2.DEFAULT_GROUP))
+
+ // The listener lifecycle lock serializes add/remove bookkeeping with
the
+ // corresponding ListenConfig/CancelListenConfig calls. This prevents a
+ // concurrent addListener from slipping in between removeListener's
emptiness
+ // check and its CancelListenConfig call.
+ n.listenerLock.Lock()
+ defer n.listenerLock.Unlock()
+
+ rawSet, loaded := n.keyListeners.Load(key)
if !loaded {
- _, cancel := context.WithCancel(context.Background())
- listenersMap := &sync.Map{}
- listenersMap.Store(listener, cancel)
+ set := newKeyListenerSet(group)
+ set.add(listener)
// double load for invalid race
- rawListenersMap, loaded = n.keyListeners.LoadOrStore(key,
listenersMap)
+ var actual any
+ actual, loaded = n.keyListeners.LoadOrStore(key, set)
if !loaded {
err := n.client.Client().ListenConfig(vo.ConfigParam{
DataId: key,
- Group:
n.resolvedGroup(n.url.GetParam(constant.NacosGroupKey,
constant2.DEFAULT_GROUP)),
+ Group: group,
OnChange: func(namespace, group, dataId, data
string) {
- go callback(listenersMap, namespace,
group, dataId, data)
+ go callback(set, namespace, group,
dataId, data)
},
})
if err != nil {
@@ -69,18 +120,38 @@ func (n *nacosDynamicConfiguration) addListener(key
string, listener config_cent
}
return
}
+ rawSet = actual
}
- _, cancel := context.WithCancel(context.Background())
- listenersMap := rawListenersMap.(*sync.Map)
- listenersMap.Store(listener, cancel)
+ rawSet.(*keyListenerSet).add(listener)
}
func (n *nacosDynamicConfiguration) removeListener(key string, listener
config_center.ConfigurationListener) {
- rawListenersMap, loaded := n.keyListeners.Load(key)
+ n.listenerLock.Lock()
+ defer n.listenerLock.Unlock()
+
+ rawSet, loaded := n.keyListeners.Load(key)
if !loaded {
logger.Errorf("[ConfigCenter][Nacos] key is not be listened,
key=%s", key)
- } else {
- listenersMap := rawListenersMap.(*sync.Map)
- listenersMap.Delete(listener)
+ return
+ }
+ set := rawSet.(*keyListenerSet)
+ isEmpty := set.remove(listener)
+
+ if isEmpty {
+ // Delete from keyListeners and cancel the nacos subscription
as a single
+ // atomic step. Because addListener takes the same lock, no
concurrent add
+ // can register a fresh subscription between this Delete and the
+ // CancelListenConfig below — the race that previously let a
late cancel
+ // drop a just-registered listener.
+ n.keyListeners.Delete(key)
+ if n.client != nil {
+ err :=
n.client.Client().CancelListenConfig(vo.ConfigParam{
+ DataId: key,
+ Group: set.group,
+ })
+ if err != nil {
+ logger.Errorf("[ConfigCenter][Nacos] cancel
listen config fail, key=%s, err=%v", key, err)
+ }
+ }
}
}
diff --git a/config_center/nacos/listener_concurrency_test.go
b/config_center/nacos/listener_concurrency_test.go
new file mode 100644
index 000000000..3b8b030fb
--- /dev/null
+++ b/config_center/nacos/listener_concurrency_test.go
@@ -0,0 +1,320 @@
+/*
+ * 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 nacos
+
+import (
+ "strconv"
+ "sync"
+ "sync/atomic"
+ "testing"
+ "time"
+)
+
+import (
+ nacosClient "github.com/dubbogo/gost/database/kv/nacos"
+
+ "github.com/nacos-group/nacos-sdk-go/v2/model"
+ "github.com/nacos-group/nacos-sdk-go/v2/vo"
+
+ "github.com/stretchr/testify/assert"
+)
+
+import (
+ "dubbo.apache.org/dubbo-go/v3/common"
+ "dubbo.apache.org/dubbo-go/v3/common/constant"
+ "dubbo.apache.org/dubbo-go/v3/config_center"
+)
+
+// noopListener is a no-op ConfigurationListener used by the concurrency tests.
+// It carries an id so each instance has a distinct address — an empty struct
+// would have Go collapse every &noopListener{} to a single address
+// (runtime.zerobase), making every listener the same map key and invalidating
+// the add/remove bookkeeping under test.
+type noopListener struct {
+ id int
+}
+
+var listenerSeq int32
+
+func newNoopListener() *noopListener {
+ return &noopListener{id: int(atomic.AddInt32(&listenerSeq, 1))}
+}
+
+func (r *noopListener) Process(*config_center.ConfigChangeEvent) {}
+
+// countingConfigClient is a concurrency-safe IConfigClient that mirrors the
+// real Nacos SDK semantics relevant to the TOCTOU race under review: both
+// ListenConfig and CancelListenConfig operate on a single per-(dataId, group)
+// slot. ListenConfig sets the slot; CancelListenConfig clears it
+// unconditionally — exactly like the SDK's cacheMap.Remove, which drops
+// whatever subscription currently occupies the slot regardless of when it was
+// registered. This is what makes the race dangerous: a late cancel can wipe a
+// subscription a racing listen just established.
+//
+// The optional onListenDone / onCancelEnter hooks let a deterministic test
+// choreograph the exact interleaving the reviewer described (a cancel landing
+// after a racing listen) without relying on timing.
+type countingConfigClient struct {
+ mu sync.Mutex
+ slot map[string]struct{} // (dataId|group) -> present when a
subscription is live
+ listens int32
+ cancels int32
+ onListenDone func() // invoked after ListenConfig sets the slot
+ onCancelEnter func() // invoked before CancelListenConfig clears the
slot
+}
+
+func newCountingConfigClient() *countingConfigClient {
+ return &countingConfigClient{slot: make(map[string]struct{})}
+}
+
+func (c *countingConfigClient) keyOf(p vo.ConfigParam) string { return
p.DataId + "|" + p.Group }
+
+func (c *countingConfigClient) ListenConfig(p vo.ConfigParam) error {
+ c.mu.Lock()
+ c.slot[c.keyOf(p)] = struct{}{}
+ c.mu.Unlock()
+ atomic.AddInt32(&c.listens, 1)
+ if c.onListenDone != nil {
+ c.onListenDone()
+ }
+ return nil
+}
+
+func (c *countingConfigClient) CancelListenConfig(p vo.ConfigParam) error {
+ if c.onCancelEnter != nil {
+ c.onCancelEnter()
+ }
+ c.mu.Lock()
+ delete(c.slot, c.keyOf(p))
+ c.mu.Unlock()
+ atomic.AddInt32(&c.cancels, 1)
+ return nil
+}
+
+// hasSubscription reports whether a live subscription occupies the slot.
+func (c *countingConfigClient) hasSubscription(key, group string) bool {
+ c.mu.Lock()
+ defer c.mu.Unlock()
+ _, ok := c.slot[key+"|"+group]
+ return ok
+}
+
+func (c *countingConfigClient) GetConfig(vo.ConfigParam) (string, error) {
return "", nil }
+func (c *countingConfigClient) PublishConfig(vo.ConfigParam) (bool, error) {
return true, nil }
+func (c *countingConfigClient) DeleteConfig(vo.ConfigParam) (bool, error) {
return true, nil }
+func (c *countingConfigClient) SearchConfig(vo.SearchConfigParam)
(*model.ConfigPage, error) {
+ return nil, nil
+}
+func (c *countingConfigClient) CloseClient() {}
+
+// newTestConfig builds a nacosDynamicConfiguration wired to a counting client.
+func newTestConfig(t *testing.T, client *countingConfigClient)
*nacosDynamicConfiguration {
+ t.Helper()
+ nc := &nacosClient.NacosConfigClient{}
+ nc.SetClient(client)
+ u, err := common.NewURL("registry://127.0.0.1:8848",
+ common.WithParamsValue(constant.NacosGroupKey, "test-group"))
+ assert.NoError(t, err)
+ return &nacosDynamicConfiguration{
+ url: u,
+ client: nc,
+ }
+}
+
+// TestAddRemoveListenerConcurrentChurnKeepsPersistentSubscription verifies the
+// non-empty listener path under concurrent churn. A persistent listener
remains
+// registered while many temporary listeners are added and removed, so the key
+// should never be canceled and the persistent subscription must remain live.
+func TestAddRemoveListenerConcurrentChurnKeepsPersistentSubscription(t
*testing.T) {
+ const key = "race-key"
+ client := newCountingConfigClient()
+ n := newTestConfig(t, client)
+
+ // A persistent listener that stays registered for the whole test; its
+ // subscription must never be dropped by a racing remove's cancel.
+ persistent := newNoopListener()
+ n.addListener(key, persistent)
+ assert.True(t, client.hasSubscription(key, "test-group"),
+ "initial add should establish a subscription")
+
+ var wg sync.WaitGroup
+ workers := 32
+ start := make(chan struct{})
+ for w := 0; w < workers; w++ {
+ wg.Add(1)
+ go func() {
+ defer wg.Done()
+ <-start
+ for i := 0; i < 200; i++ {
+ l := newNoopListener()
+ n.addListener(key, l)
+ n.removeListener(key, l)
+ }
+ }()
+ }
+ close(start)
+ wg.Wait()
+
+ // The persistent listener is still registered, so a live subscription
must
+ // remain. Under the unfixed code a racing cancel could have cleared
the slot
+ // after a worker's add repopulated it, leaving the slot empty while the
+ // persistent listener is still registered — the assertion below catches
+ // that. (Each worker balances its own add/remove, so no worker listener
+ // should remain; only the persistent one.)
+ assert.True(t, client.hasSubscription(key, "test-group"),
+ "persistent listener's subscription must survive concurrent
churn")
+ assert.Equal(t, int32(0), atomic.LoadInt32(&client.cancels),
+ "persistent listener keeps the key non-empty, so this test
should not exercise cancel")
+
+ // Sanity: the persistent listener is still the only one tracked for
the key.
+ rawSet, ok := n.keyListeners.Load(key)
+ assert.True(t, ok, "key entry must remain while a listener is
registered")
+ set := rawSet.(*keyListenerSet)
+ assert.Len(t, set.snapshot(), 1, "only the persistent listener should
remain")
+}
+
+// TestRemoveThenAddNoLostSubscriptionDeterministic forces the exact
interleaving
+// the reviewer described, with no reliance on timing:
+//
+// 1. remover: removeListener observes the last listener leaving, deletes the
+// keyListeners entry, and enters CancelListenConfig — where we park it.
+// 2. adder: while the remover is parked inside CancelListenConfig,
addListener
+// runs: it sees no entry, stores a fresh set, and calls ListenConfig
(which
+// populates the slot).
+// 3. remover: we release CancelListenConfig; it clears the slot
unconditionally
+// (mirroring the SDK's cacheMap.Remove).
+//
+// Under the unfixed code the adder's slot is wiped by the late cancel,
leaving a
+// registered listener with no live Nacos subscription. Under the fix the
+// listener lifecycle lock makes step 2 wait until the remover has fully
finished
+// (including its CancelListenConfig), so the adder's ListenConfig wins and the
+// slot survives.
+func TestRemoveThenAddNoLostSubscriptionDeterministic(t *testing.T) {
+ const key = "det-key"
+
+ // cancelEntered is closed once the remover is parked inside
CancelListenConfig;
+ // cancelRelease unblocks it to finish clearing the slot.
+ cancelEntered := make(chan struct{})
+ cancelRelease := make(chan struct{})
+
+ client := newCountingConfigClient()
+ client.onCancelEnter = func() {
+ close(cancelEntered)
+ <-cancelRelease
+ }
+ n := newTestConfig(t, client)
+
+ // One registered listener so removeListener has a real last-listener
to remove.
+ first := newNoopListener()
+ n.addListener(key, first)
+ assert.True(t, client.hasSubscription(key, "test-group"))
+
+ removerDone := make(chan struct{})
+ go func() {
+ defer close(removerDone)
+ n.removeListener(key, first)
+ }()
+
+ // Wait until the remover is parked inside CancelListenConfig (i.e. it
has
+ // already deleted the keyListeners entry). This is the dangerous
window.
+ <-cancelEntered
+
+ // Now the adder races in: under the fix it blocks on the listener
lifecycle
+ // lock held by the remover, so ListenConfig has NOT run yet. Under the
+ // unfixed code the adder proceeds, stores a fresh set, and ListenConfig
+ // populates the slot while the remover is still parked.
+ var adderWG sync.WaitGroup
+ adderWG.Add(1)
+ go func() {
+ defer adderWG.Done()
+ l := newNoopListener()
+ n.addListener(key, l)
+ }()
+
+ // Probe whether the adder could complete before we release the cancel.
With
+ // the lifecycle lock the adder is still blocked; without it the adder
+ // finished and populated the slot. Either way we then release the
cancel and
+ // wait for both goroutines to finish.
+ time.Sleep(200 * time.Millisecond)
+ close(cancelRelease)
+
+ <-removerDone
+ adderWG.Wait()
+
+ // The adder registered a listener, so a live subscription must remain.
On the
+ // unfixed code the late cancel wiped the adder's slot, leaving it
empty —
+ // this assertion fails there and passes with the fix.
+ assert.True(t, client.hasSubscription(key, "test-group"),
+ "adder's subscription must survive the racing remover's cancel")
+ rawSet, ok := n.keyListeners.Load(key)
+ assert.True(t, ok, "key entry must remain while the adder's listener is
registered")
+ assert.Len(t, rawSet.(*keyListenerSet).snapshot(), 1, "only the adder's
listener should remain")
+}
+
+// TestListenerConcurrentStress runs many goroutines adding and removing
+// listeners across several keys. Run with -race to catch data races. The
+// invariant checked at the end: for every key, a live Nacos subscription
+// exists iff at least one listener remains registered. Under the unfixed code
+// the TOCTOU window can leave a key with registered listeners but no slot
+// (or, less likely, an orphaned slot after all listeners are gone).
+func TestListenerConcurrentStress(t *testing.T) {
+ client := newCountingConfigClient()
+ n := newTestConfig(t, client)
+
+ const numKeys = 8
+ const goroutines = 64
+ const iterations = 200
+
+ // remaining[k] is the number of listeners we intentionally left
registered
+ // for key k (those we did not remove).
+ var remaining [numKeys]int32
+
+ keyFor := func(k int) string { return "stress-key-" + strconv.Itoa(k) }
+
+ var wg sync.WaitGroup
+ for g := 0; g < goroutines; g++ {
+ wg.Add(1)
+ go func(g int) {
+ defer wg.Done()
+ for i := 0; i < iterations; i++ {
+ k := g % numKeys
+ key := keyFor(k)
+ l := newNoopListener()
+ n.addListener(key, l)
+ // Keep roughly half of the listeners so some
keys retain
+ // subscriptions and others churn to empty
(exercising the
+ // cancel path) and back.
+ if (i+g)%2 == 0 {
+ n.removeListener(key, l)
+ } else {
+ atomic.AddInt32(&remaining[k], 1)
+ }
+ }
+ }(g)
+ }
+ wg.Wait()
+
+ // A live subscription must exist iff at least one listener remains.
+ for k := 0; k < numKeys; k++ {
+ key := keyFor(k)
+ got := client.hasSubscription(key, "test-group")
+ want := atomic.LoadInt32(&remaining[k]) > 0
+ assert.Equal(t, want, got,
+ "key %s: subscription live=%v but remaining
listeners=%d", key, got, remaining[k])
+ }
+}
diff --git a/config_center/nacos/listener_test.go
b/config_center/nacos/listener_test.go
index 00ceb7f1d..121fe5d32 100644
--- a/config_center/nacos/listener_test.go
+++ b/config_center/nacos/listener_test.go
@@ -18,7 +18,6 @@
package nacos
import (
- "sync"
"testing"
)
@@ -37,10 +36,10 @@ func (r *recordingListener) Process(e
*config_center.ConfigChangeEvent) {
func TestCallback(t *testing.T) {
l := &recordingListener{}
- var m sync.Map
- m.Store(l, struct{}{})
+ set := newKeyListenerSet("test-group")
+ set.add(l)
- callback(&m, "", "g", "data", "payload")
+ callback(set, "", "g", "data", "payload")
if len(l.events) != 1 {
t.Fatalf("expected 1 event, got %d", len(l.events))
@@ -54,13 +53,40 @@ func TestRemoveListener(t *testing.T) {
n := &nacosDynamicConfiguration{}
key := "k"
l := &recordingListener{}
- inner := &sync.Map{}
- inner.Store(l, struct{}{})
- n.keyListeners.Store(key, inner)
+ set := newKeyListenerSet("test-group")
+ set.add(l)
+ n.keyListeners.Store(key, set)
n.removeListener(key, l)
- if _, ok := inner.Load(l); ok {
+ if _, ok := set.listeners[l]; ok {
t.Fatalf("listener should be removed")
}
+ // After removing the only listener, the key should be deleted from
keyListeners
+ if _, loaded := n.keyListeners.Load(key); loaded {
+ t.Fatalf("key should be deleted from keyListeners after last
listener is removed")
+ }
+}
+
+func TestRemoveListenerMultipleListeners(t *testing.T) {
+ n := &nacosDynamicConfiguration{}
+ key := "k"
+ l1 := &recordingListener{}
+ l2 := &recordingListener{}
+ set := newKeyListenerSet("test-group")
+ set.add(l1)
+ set.add(l2)
+ n.keyListeners.Store(key, set)
+
+ // Remove first listener — key should still exist
+ n.removeListener(key, l1)
+ if _, loaded := n.keyListeners.Load(key); !loaded {
+ t.Fatalf("key should still exist after removing one of multiple
listeners")
+ }
+
+ // Remove second listener — key should be deleted
+ n.removeListener(key, l2)
+ if _, loaded := n.keyListeners.Load(key); loaded {
+ t.Fatalf("key should be deleted from keyListeners after last
listener is removed")
+ }
}
