fix(zookeeper): align listener paths and empty config events

Author: ywxzm03

config_center/zookeeper/impl.go

@@ -110,10 +110,8 @@ func newZookeeperDynamicConfiguration(url *common.URL) (*zookeeperDynamicConfigu
 }
 
 // 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)
 }
 
@@ -128,8 +126,7 @@ func buildPath(rootPath, subPath string) string {
 }
 
 func (c *zookeeperDynamicConfiguration) RemoveListener(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.RemoveListener(qualifiedKey, listener)
 }
 

config_center/zookeeper/impl_test.go

@@ -183,6 +183,37 @@ func TestLoadPropertiesRegistersWatchOnlyWhenInactive(t *testing.T) {
 	require.True(t, waitForEvent(inactivePath, time.Second))
 }
 
+func TestListenerUsesGroupOption(t *testing.T) {
+	cluster, client, _, err := gxzookeeper.NewMockZookeeperClient("listener-group", 5*time.Second)
+	if err != nil {
+		t.Skipf("skip mock zk setup: %v", err)
+	}
+	defer cluster.Stop()
+
+	zkListener := remotingzookeeper.NewZkEventListener(client)
+	defer zkListener.Close()
+	cfg := &zookeeperDynamicConfiguration{
+		rootPath: "/dubbo/config",
+		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) {
 	cluster, client, events, err := gxzookeeper.NewMockZookeeperClient("watch-limit", 5*time.Second)
 	if err != nil {

config_center/zookeeper/listener.go

@@ -203,19 +203,14 @@ func (l *CacheListener) DataChange(event remoting.Event) bool {
 		}
 	}
 
-	changeType := event.Action
-	if event.Content == "" {
-		changeType = remoting.EventTypeDel
-	}
-
 	key, group := l.pathToKeyGroup(event.Path)
-	defer metrics.Publish(metricsConfigCenter.NewIncMetricEvent(key, group, changeType, metricsConfigCenter.Zookeeper))
+	defer metrics.Publish(metricsConfigCenter.NewIncMetricEvent(key, group, event.Action, 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,
+				ConfigType: event.Action,
 			})
 		}
 		return true

config_center/zookeeper/listener_test.go

@@ -69,7 +69,7 @@ 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)
@@ -78,6 +78,9 @@ func TestCacheListenerDataChangeEmptyContent(t *testing.T) {
 	}
 
 	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)