Fix:zk too many tcp conn

cluster/router/chain/chain_test.go

@@ -25,6 +25,7 @@ import (
 )
 
 import (
+	zk "github.com/dubbogo/go-zookeeper/zk"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -53,16 +54,19 @@ const (
 	consumerFormat   = "consumer://%s/com.foo.BarService"
 	dubboForamt      = "dubbo://%s:%d/com.foo.BarService"
 	anyUrlFormat     = "condition://%s/com.foo.BarService"
-	zk               = "zookeeper"
+	zkName           = "zookeeper"
 	applicationKey   = "test-condition"
 	applicationField = "application"
 	forceField       = "force"
 	forceValue       = "true"
 )
 
+var zkCluster *zk.TestCluster
+
 func TestNewRouterChain(t *testing.T) {
 	ts, z, _, err := zookeeper.NewMockZookeeperClient("test", 15*time.Second)
 	assert.NoError(t, err)
+	zkCluster = ts
 	err = z.Create(path)
 	assert.NoError(t, err)
 	testyml := `scope: application
@@ -77,12 +81,12 @@ conditions:
 	_, err = z.Conn.Set(path, []byte(testyml), 0)
 	assert.NoError(t, err)
 	defer func() {
-		_ = ts.Stop()
+		z.Delete(path)
 		z.Close()
 	}()
 
 	zkUrl, _ := common.NewURL(fmt.Sprintf(zkFormat, localIP, ts.Servers[0].Port))
-	configuration, err := extension.GetConfigCenterFactory(zk).GetDynamicConfiguration(zkUrl)
+	configuration, err := extension.GetConfigCenterFactory(zkName).GetDynamicConfiguration(zkUrl)
 	config.GetEnvInstance().SetDynamicConfiguration(configuration)
 
 	assert.Nil(t, err)
@@ -114,7 +118,7 @@ func TestNewRouterChainURLNil(t *testing.T) {
 }
 
 func TestRouterChainAddRouters(t *testing.T) {
-	ts, z, _, err := zookeeper.NewMockZookeeperClient("test", 15*time.Second)
+	_, z, _, err := zookeeper.NewMockZookeeperClient("test", 15*time.Second, zookeeper.WithTestCluster(zkCluster))
 	assert.NoError(t, err)
 	err = z.Create(path)
 	assert.NoError(t, err)
@@ -131,13 +135,12 @@ conditions:
 	_, err = z.Conn.Set(path, []byte(testyml), 0)
 	assert.NoError(t, err)
 	defer func() {
-		_ = ts.Stop()
-		assert.NoError(t, err)
+		z.Delete(path)
 		z.Close()
 	}()
 
-	zkUrl, _ := common.NewURL(fmt.Sprintf(zkFormat, localIP, ts.Servers[0].Port))
-	configuration, err := extension.GetConfigCenterFactory(zk).GetDynamicConfiguration(zkUrl)
+	zkUrl, _ := common.NewURL(fmt.Sprintf(zkFormat, localIP, zkCluster.Servers[0].Port))
+	configuration, err := extension.GetConfigCenterFactory(zkName).GetDynamicConfiguration(zkUrl)
 	assert.NoError(t, err)
 	config.GetEnvInstance().SetDynamicConfiguration(configuration)
 
@@ -164,15 +167,10 @@ conditions:
 }
 
 func TestRouterChainRoute(t *testing.T) {
-	ts, z, _, err := zookeeper.NewMockZookeeperClient("test", 15*time.Second)
-	defer func() {
-		err = ts.Stop()
-		assert.NoError(t, err)
-		z.Close()
-	}()
-
+	ts, _, _, err := zookeeper.NewMockZookeeperClient("test", 15*time.Second, zookeeper.WithTestCluster(zkCluster))
+	assert.Nil(t, err)
 	zkUrl, _ := common.NewURL(fmt.Sprintf(zkFormat, localIP, ts.Servers[0].Port))
-	configuration, err := extension.GetConfigCenterFactory(zk).GetDynamicConfiguration(zkUrl)
+	configuration, err := extension.GetConfigCenterFactory(zkName).GetDynamicConfiguration(zkUrl)
 	assert.NoError(t, err)
 	config.GetEnvInstance().SetDynamicConfiguration(configuration)
 
@@ -197,7 +195,7 @@ func TestRouterChainRoute(t *testing.T) {
 }
 
 func TestRouterChainRouteAppRouter(t *testing.T) {
-	ts, z, _, err := zookeeper.NewMockZookeeperClient("test", 15*time.Second)
+	ts, z, _, err := zookeeper.NewMockZookeeperClient("test", 15*time.Second, zookeeper.WithTestCluster(zkCluster))
 	assert.NoError(t, err)
 	err = z.Create(path)
 	assert.NoError(t, err)
@@ -214,13 +212,12 @@ conditions:
 	_, err = z.Conn.Set(path, []byte(testyml), 0)
 	assert.NoError(t, err)
 	defer func() {
-		_ = ts.Stop()
-		assert.NoError(t, err)
+		z.Delete(path)
 		z.Close()
 	}()
 
 	zkUrl, _ := common.NewURL(fmt.Sprintf(zkFormat, localIP, ts.Servers[0].Port))
-	configuration, err := extension.GetConfigCenterFactory(zk).GetDynamicConfiguration(zkUrl)
+	configuration, err := extension.GetConfigCenterFactory(zkName).GetDynamicConfiguration(zkUrl)
 	assert.NoError(t, err)
 	config.GetEnvInstance().SetDynamicConfiguration(configuration)
 
@@ -242,7 +239,7 @@ conditions:
 }
 
 func TestRouterChainRouteNoRoute(t *testing.T) {
-	ts, z, _, err := zookeeper.NewMockZookeeperClient("test", 15*time.Second)
+	ts, z, _, err := zookeeper.NewMockZookeeperClient("test", 15*time.Second, zookeeper.WithTestCluster(zkCluster))
 	assert.Nil(t, err)
 	defer func() {
 		_ = ts.Stop()
@@ -251,7 +248,7 @@ func TestRouterChainRouteNoRoute(t *testing.T) {
 	}()
 
 	zkUrl, _ := common.NewURL(fmt.Sprintf(zkFormat, localIP, ts.Servers[0].Port))
-	configuration, err := extension.GetConfigCenterFactory(zk).GetDynamicConfiguration(zkUrl)
+	configuration, err := extension.GetConfigCenterFactory(zkName).GetDynamicConfiguration(zkUrl)
 	config.GetEnvInstance().SetDynamicConfiguration(configuration)
 	assert.Nil(t, err)
 

cluster/router/condition/app_router_test.go

@@ -68,7 +68,6 @@ conditions:
 	_, err = z.Conn.Set(routerPath, []byte(testYML), 0)
 	assert.NoError(t, err)
 	defer func() {
-		err = ts.Stop()
 		assert.NoError(t, err)
 		z.Close()
 	}()
@@ -124,7 +123,6 @@ conditions:
 	_, err = z.Conn.Set(routerPath, []byte(testYML), 0)
 	assert.NoError(t, err)
 	defer func() {
-		err = ts.Stop()
 		assert.NoError(t, err)
 		z.Close()
 	}()

cluster/router/local/factory.go

@@ -38,6 +38,6 @@ func newLocalPriorityRouteFactory() router.PriorityRouterFactory {
 }
 
 // NewPriorityRouter construct a new NewLocalDiscRouter via url
-func (f *LocalPriorityRouteFactory) NewPriorityRouter(url *common.URL) (router.PriorityRouter, error) {
+func (f *LocalPriorityRouteFactory) NewPriorityRouter(url *common.URL, ch chan struct{}) (router.PriorityRouter, error) {
 	return NewLocalPriorityRouter(url)
 }

cluster/router/tag/tag_router_test.go

@@ -334,8 +334,6 @@ tags:
 
 func (suite *DynamicTagRouter) TearDownTest() {
 	suite.zkClient.Close()
-	err := suite.testCluster.Stop()
-	suite.Nil(err)
 }
 
 func (suite *DynamicTagRouter) TestDynamicTagRouterSetByIPv4() {

config_center/zookeeper/impl.go

@@ -206,10 +206,9 @@ func (c *zookeeperDynamicConfiguration) IsAvailable() bool {
 }
 
 func (c *zookeeperDynamicConfiguration) closeConfigs() {
+	logger.Infof("begin to close provider zk client")
 	c.cltLock.Lock()
 	defer c.cltLock.Unlock()
-	logger.Infof("begin to close provider zk client")
-	// Close the old client first to close the tmp node
 	c.client.Close()
 	c.client = nil
 }

config_center/zookeeper/impl_test.go

@@ -81,10 +81,10 @@ func initZkData(group string, t *testing.T) (*zk.TestCluster, *zookeeperDynamicC
 	dubbo.service.com.ikurento.user.UserProvider.cluster=failover
 `
 	if group != "" {
-		err = zreg.client.Create(path.Join(zreg.rootPath, "dubbo", dubboPropertyFileName))
+		err = zreg.client.Create(path.Join(zreg.rootPath, group, dubboPropertyFileName))
 		assert.NoError(t, err)
 
-		_, err = zreg.client.Conn.Set(path.Join(zreg.rootPath, "dubbo", dubboPropertyFileName), []byte(data), 0)
+		_, err = zreg.client.Conn.Set(path.Join(zreg.rootPath, group, dubboPropertyFileName), []byte(data), 0)
 		assert.NoError(t, err)
 	} else {
 		err = zreg.client.Create(path.Join(zreg.rootPath, dubboPropertyFileName))
@@ -100,6 +100,7 @@ func initZkData(group string, t *testing.T) (*zk.TestCluster, *zookeeperDynamicC
 func TestGetConfig(t *testing.T) {
 	ts, reg := initZkData("dubbo", t)
 	defer func() {
+		reg.client.Close()
 		err := ts.Stop()
 		assert.NoError(t, err)
 	}()
@@ -122,11 +123,13 @@ func TestGetConfig(t *testing.T) {
 func TestAddListener(t *testing.T) {
 	ts, reg := initZkData("", t)
 	defer func() {
+		reg.client.Close()
 		err := ts.Stop()
 		assert.NoError(t, err)
 	}()
 	listener := &mockDataListener{}
 	reg.AddListener(dubboPropertyFileName, listener)
+
 	listener.wg.Add(1)
 	data := `
 	dubbo.consumer.request_timeout=3s
@@ -158,6 +161,7 @@ func TestAddListener(t *testing.T) {
 func TestRemoveListener(t *testing.T) {
 	ts, reg := initZkData("", t)
 	defer func() {
+		reg.client.Close()
 		err := ts.Stop()
 		assert.NoError(t, err)
 	}()
@@ -197,19 +201,20 @@ func TestZookeeperDynamicConfigurationPublishConfig(t *testing.T) {
 	value := "Test Data"
 	customGroup := "Custom Group"
 	key := "myKey"
-	ts, zk := initZkData(config_center.DEFAULT_GROUP, t)
+	ts, reg := initZkData(config_center.DEFAULT_GROUP, t)
 	defer func() {
+		reg.client.Close()
 		err := ts.Stop()
 		assert.NoError(t, err)
 	}()
-	err := zk.PublishConfig(key, customGroup, value)
+	err := reg.PublishConfig(key, customGroup, value)
 	assert.Nil(t, err)
-	result, err := zk.GetInternalProperty("myKey", config_center.WithGroup(customGroup))
+	result, err := reg.GetInternalProperty("myKey", config_center.WithGroup(customGroup))
 	assert.Nil(t, err)
 	assert.Equal(t, value, result)
 
 	var keys *gxset.HashSet
-	keys, err = zk.GetConfigKeysByGroup(customGroup)
+	keys, err = reg.GetConfigKeysByGroup(customGroup)
 	assert.Nil(t, err)
 	assert.Equal(t, 1, keys.Size())
 	assert.True(t, keys.Contains(key))

registry/kubernetes/registry.go

@@ -25,7 +25,7 @@ import (
 )
 
 import (
-	"github.com/apache/dubbo-getty"
+	getty "github.com/apache/dubbo-getty"
 	perrors "github.com/pkg/errors"
 	v1 "k8s.io/api/core/v1"
 )
@@ -202,7 +202,7 @@ func (r *kubernetesRegistry) HandleClientRestart() {
 		failTimes int
 	)
 
-	defer r.WaitGroup()
+	defer r.WaitGroup().Done()
 LOOP:
 	for {
 		select {

registry/zookeeper/listener.go

@@ -56,12 +56,13 @@ func (l *RegistryDataListener) SubscribeURL(url *common.URL, listener config_cen
 	l.subscribed[url.ServiceKey()] = listener
 }
 
-// UnSubscribeURL is used to set a watch listener for url
+// UnSubscribeURL is used to unset a watch listener for url
 func (l *RegistryDataListener) UnSubscribeURL(url *common.URL) config_center.ConfigurationListener {
 	if l.closed {
 		return nil
 	}
 	listener := l.subscribed[url.ServiceKey()]
+	listener.(*RegistryConfigurationListener).Close()
 	delete(l.subscribed, url.ServiceKey())
 	return listener
 }
@@ -142,9 +143,6 @@ func (l *RegistryConfigurationListener) Process(configType *config_center.Config
 func (l *RegistryConfigurationListener) Next() (*registry.ServiceEvent, error) {
 	for {
 		select {
-		case <-l.client.Done():
-			logger.Warnf("listener's zk client connection (address {%s}) is broken, so zk event listener exit now.", l.client.ZkAddrs)
-			return nil, perrors.New("zookeeper client stopped")
 		case <-l.close:
 			return nil, perrors.New("listener have been closed")
 		case <-l.registry.Done():
@@ -156,9 +154,6 @@ func (l *RegistryConfigurationListener) Next() (*registry.ServiceEvent, error) {
 				logger.Warnf("update @result{%s}. But its connection to registry is invalid", e.Value)
 				continue
 			}
-			//r.update(e.res)
-			//write to invoker
-			//r.outerEventCh <- e.res
 			return &registry.ServiceEvent{Action: e.ConfigType, Service: e.Value.(*common.URL)}, nil
 		}
 	}

registry/zookeeper/registry.go

@@ -129,7 +129,7 @@ func (r *zkRegistry) InitListeners() {
 		defer oldDataListener.mutex.Unlock()
 		r.dataListener.closed = true
 		recovered := r.dataListener.subscribed
-		if len(recovered) > 0 {
+		if recovered != nil && len(recovered) > 0 {
 			// recover all subscribed url
 			for _, oldListener := range recovered {
 				var (
@@ -282,11 +282,7 @@ func (r *zkRegistry) getListener(conf *common.URL) (*RegistryConfigurationListen
 	//Interested register to dataconfig.
 	r.dataListener.SubscribeURL(conf, zkListener)
 
-	go r.listener.ListenServiceEvent(
-		conf,
-		fmt.Sprintf("/dubbo/%s/"+constant.DEFAULT_CATEGORY, url.QueryEscape(conf.Service())),
-		r.dataListener,
-	)
+	go r.listener.ListenServiceEvent(conf, fmt.Sprintf("/dubbo/%s/"+constant.DEFAULT_CATEGORY, url.QueryEscape(conf.Service())), r.dataListener)
 
 	return zkListener, nil
 }
@@ -297,9 +293,9 @@ func (r *zkRegistry) getCloseListener(conf *common.URL) (*RegistryConfigurationL
 	r.dataListener.mutex.Lock()
 	configurationListener := r.dataListener.subscribed[conf.ServiceKey()]
 	if configurationListener != nil {
-		rcListener, _ := configurationListener.(*RegistryConfigurationListener)
-		if rcListener != nil {
-			if rcListener.isClosed {
+		zkListener, _ = configurationListener.(*RegistryConfigurationListener)
+		if zkListener != nil {
+			if zkListener.isClosed {
 				r.dataListener.mutex.Unlock()
 				return nil, perrors.New("configListener already been closed")
 			}

registry/zookeeper/registry_test.go

@@ -37,11 +37,14 @@ func Test_Register(t *testing.T) {
 	regURL, _ := common.NewURL("registry://127.0.0.1:1111", common.WithParamsValue(constant.ROLE_KEY, strconv.Itoa(common.PROVIDER)))
 	url, _ := common.NewURL("dubbo://127.0.0.1:20000/com.ikurento.user.UserProvider", common.WithParamsValue(constant.CLUSTER_KEY, "mock"), common.WithParamsValue("serviceid", "soa.mock"), common.WithMethods([]string{"GetUser", "AddUser"}))
 
-	ts, reg, _ := newMockZkRegistry(regURL)
+	ts, reg, err := newMockZkRegistry(regURL)
+	if err != nil {
+		assert.NoError(t, err)
+	}
 	defer func() {
 		_ = ts.Stop()
 	}()
-	err := reg.Register(url)
+	err = reg.Register(url)
 	children, _ := reg.client.GetChildren("/dubbo/com.ikurento.user.UserProvider/providers")
 	assert.Regexp(t, ".*dubbo%3A%2F%2F127.0.0.1%3A20000%2Fcom.ikurento.user.UserProvider%3Fanyhost%3Dtrue%26cluster%3Dmock%26.*.serviceid%3Dsoa.mock", children)
 	assert.NoError(t, err)

registry/zookeeper/service_discovery.go

@@ -24,10 +24,9 @@ import (
 	"strings"
 	"sync"
 )
-
 import (
-	"github.com/dubbogo/gost/container/set"
-	"github.com/dubbogo/gost/hash/page"
+	gxset "github.com/dubbogo/gost/container/set"
+	gxpage "github.com/dubbogo/gost/hash/page"
 	perrors "github.com/pkg/errors"
 )
 
@@ -111,6 +110,7 @@ func newZookeeperServiceDiscovery(name string) (registry.ServiceDiscovery, error
 	if err != nil {
 		return nil, err
 	}
+	zksd.WaitGroup().Add(1) //zk client start successful, then wg +1
 	go zookeeper.HandleClientRestart(zksd)
 	zksd.csd = curator_discovery.NewServiceDiscovery(zksd.client, rootPath)
 	return zksd, nil