Fix 3.0 ci problem (#1188)

* fix etcd problem

* fix metadata report etcd problem

* fix nacos & cluster_impl ci problem

* fix nacos ci problem

* fix zk ci problem

* fix zk metadata service urls not found error

* fix zk TestAddListenerZookeeperServiceDiscovery

* fix zk TestAddListenerZookeeperServiceDiscovery

* fix config int to string bug

cluster/cluster_impl/failback_cluster_invoker.go

@@ -169,7 +169,7 @@ func (invoker *failbackClusterInvoker) Invoke(ctx context.Context, invocation pr
 		logger.Errorf("Failback to invoke the method %v in the service %v, wait for retry in background. Ignored exception: %v.\n",
 			methodName, url.Service(), result.Error().Error())
 		// ignore
-		return &protocol.RPCResult{Err: result.Error()}
+		return &protocol.RPCResult{}
 	}
 	return result
 }

config/config_loader.go

@@ -109,7 +109,7 @@ func setDefaultValue(target interface{}) {
 		if len(p.Protocols) == 0 {
 			p.Protocols[constant.DEFAULT_PROTOCOL] = &ProtocolConfig{
 				Name: constant.DEFAULT_PROTOCOL,
-				Port: string(constant.DEFAULT_PORT),
+				Port: strconv.Itoa(constant.DEFAULT_PORT),
 			}
 		}
 		if p.ApplicationConfig == nil {

go.mod

@@ -12,11 +12,10 @@ require (
 	github.com/alibaba/sentinel-golang v1.0.2
 	github.com/apache/dubbo-getty v1.4.3
 	github.com/apache/dubbo-go-hessian2 v1.9.1
-	github.com/coreos/etcd v3.3.25+incompatible
 	github.com/creasty/defaults v1.5.1
 	github.com/docker/go-connections v0.4.0 // indirect
 	github.com/dubbogo/go-zookeeper v1.0.3
-	github.com/dubbogo/gost v1.11.7
+	github.com/dubbogo/gost v1.11.8
 	github.com/dubbogo/triple v0.1.3
 	github.com/elazarl/go-bindata-assetfs v1.0.0 // indirect
 	github.com/emicklei/go-restful/v3 v3.4.0
@@ -51,6 +50,7 @@ require (
 	github.com/stretchr/objx v0.2.0 // indirect
 	github.com/stretchr/testify v1.7.0
 	github.com/zouyx/agollo/v3 v3.4.5
+	go.etcd.io/etcd v0.0.0-20200402134248-51bdeb39e698
 	go.uber.org/atomic v1.7.0
 	go.uber.org/zap v1.16.0
 	google.golang.org/grpc v1.33.1

go.sum

@@ -59,7 +59,6 @@ github.com/NYTimes/gziphandler v1.1.1/go.mod h1:n/CVRwUEOgIxrgPvAQhUUr9oeUtvrhMo
 github.com/OneOfOne/xxhash v1.2.2/go.mod h1:HSdplMjZKSmBqAxg5vPj2TmRDmfkzw+cTzAElWljhcU=
 github.com/PuerkitoBio/purell v1.0.0/go.mod h1:c11w/QuzBsJSee3cPx9rAFu61PvFxuPbtSwDGJws/X0=
 github.com/PuerkitoBio/urlesc v0.0.0-20160726150825-5bd2802263f2/go.mod h1:uGdkoq3SwY9Y+13GIhn11/XLaGBb4BfwItxLd5jeuXE=
-github.com/RoaringBitmap/roaring v0.5.5/go.mod h1:puNo5VdzwbaIQxSiDIwfXl4Hnc+fbovcX4IW/dSTtUk=
 github.com/RoaringBitmap/roaring v0.6.0 h1:tZcn2nJpUrZf+xQY8x+9QY7BxSETMjkdNG4Ts5zahyU=
 github.com/RoaringBitmap/roaring v0.6.0/go.mod h1:WZ83fjBF/7uBHi6QoFyfGL4+xuV4Qn+xFkm4+vSzrhE=
 github.com/Shopify/sarama v1.19.0/go.mod h1:FVkBWblsNy7DGZRfXLU0O9RCGt5g3g3yEuWXgklEdEo=
@@ -139,8 +138,6 @@ github.com/coreos/bbolt v1.3.4 h1:0VqjxUwoTLxM3PmsSIk0hI2ao6gTtButQ2z8FT4//yo=
 github.com/coreos/bbolt v1.3.4/go.mod h1:G5EMThwa9y8QZGBClrRx5EY+Yw9kAhnjy3bSjsnlVTQ=
 github.com/coreos/etcd v3.3.10+incompatible/go.mod h1:uF7uidLiAD3TWHmW31ZFd/JWoc32PjwdhPthX9715RE=
 github.com/coreos/etcd v3.3.13+incompatible/go.mod h1:uF7uidLiAD3TWHmW31ZFd/JWoc32PjwdhPthX9715RE=
-github.com/coreos/etcd v3.3.25+incompatible h1:0GQEw6h3YnuOVdtwygkIfJ+Omx0tZ8/QkVyXI4LkbeY=
-github.com/coreos/etcd v3.3.25+incompatible/go.mod h1:uF7uidLiAD3TWHmW31ZFd/JWoc32PjwdhPthX9715RE=
 github.com/coreos/go-etcd v2.0.0+incompatible/go.mod h1:Jez6KQU2B/sWsbdaef3ED8NzMklzPG4d5KIOhIy30Tk=
 github.com/coreos/go-oidc v2.1.0+incompatible h1:sdJrfw8akMnCuUlaZU3tE/uYXFgfqom8DBE9so9EBsM=
 github.com/coreos/go-oidc v2.1.0+incompatible/go.mod h1:CgnwVTmzoESiwO9qyAFEMiHoZ1nMCKZlZ9V6mm3/LKc=
@@ -186,8 +183,8 @@ github.com/dubbogo/go-zookeeper v1.0.3 h1:UkuY+rBsxdT7Bs63QAzp9z7XqQ53W1j8E5rwl8
 github.com/dubbogo/go-zookeeper v1.0.3/go.mod h1:fn6n2CAEer3novYgk9ULLwAjuV8/g4DdC2ENwRb6E+c=
 github.com/dubbogo/gost v1.9.0/go.mod h1:pPTjVyoJan3aPxBPNUX0ADkXjPibLo+/Ib0/fADXSG8=
 github.com/dubbogo/gost v1.10.1/go.mod h1:+mQGS51XQEUWZP2JeGZTxJwipjRKtJO7Tr+FOg+72rI=
-github.com/dubbogo/gost v1.11.7 h1:mS2nuUOhOQmHSPnbs/94GakMigcKhzRr0TFLreiqNyo=
-github.com/dubbogo/gost v1.11.7/go.mod h1:2nB8jSrxVPwW5DBsRu3FZQH1+Ual3wnRHwFqjG9+4PY=
+github.com/dubbogo/gost v1.11.8 h1:OPTG4qIyNQ949GbdgHvpvYiVNno/X/YBozOVBLuNkS4=
+github.com/dubbogo/gost v1.11.8/go.mod h1:2nB8jSrxVPwW5DBsRu3FZQH1+Ual3wnRHwFqjG9+4PY=
 github.com/dubbogo/jsonparser v1.0.1/go.mod h1:tYAtpctvSP/tWw4MeelsowSPgXQRVHHWbqL6ynps8jU=
 github.com/dubbogo/net v0.0.2-0.20210326124702-e6a866993192 h1:CBEicrrVwR6u8ty+kL68ItxXVk1jaVYThrsx5ARhxUc=
 github.com/dubbogo/net v0.0.2-0.20210326124702-e6a866993192/go.mod h1:B6/ka3g8VzcyrmdCH4VkHP1K0aHeI37FmclS+TCwIBU=
@@ -798,7 +795,6 @@ github.com/tencentcloud/tencentcloud-sdk-go v3.0.83+incompatible/go.mod h1:0PfYo
 github.com/tent/http-link-go v0.0.0-20130702225549-ac974c61c2f9/go.mod h1:RHkNRtSLfOK7qBTHaeSX1D6BNpI3qw7NTxsmNr4RvN8=
 github.com/tevid/gohamcrest v1.1.1 h1:ou+xSqlIw1xfGTg1uq1nif/htZ2S3EzRqLm2BP+tYU0=
 github.com/tevid/gohamcrest v1.1.1/go.mod h1:3UvtWlqm8j5JbwYZh80D/PVBt0mJ1eJiYgZMibh0H/k=
-github.com/tinylib/msgp v1.1.0/go.mod h1:+d+yLhGm8mzTaHzB+wgMYrodPfmZrzkirds8fDWklFE=
 github.com/tmc/grpc-websocket-proxy v0.0.0-20170815181823-89b8d40f7ca8/go.mod h1:ncp9v5uamzpCO7NfCPTXjqaC+bZgJeR0sMTm6dMHP7U=
 github.com/tmc/grpc-websocket-proxy v0.0.0-20190109142713-0ad062ec5ee5/go.mod h1:ncp9v5uamzpCO7NfCPTXjqaC+bZgJeR0sMTm6dMHP7U=
 github.com/tmc/grpc-websocket-proxy v0.0.0-20201229170055-e5319fda7802 h1:uruHq4dN7GR16kFc5fp3d1RIYzJW5onx8Ybykw2YQFA=

metadata/report/etcd/report_test.go

@@ -25,8 +25,8 @@ import (
 )
 
 import (
-	"github.com/coreos/etcd/embed"
 	"github.com/stretchr/testify/assert"
+	"go.etcd.io/etcd/embed"
 )
 
 import (

registry/etcdv3/listener_test.go

@@ -25,8 +25,8 @@ import (
 
 import (
 	"github.com/apache/dubbo-getty"
-	"github.com/coreos/etcd/embed"
 	"github.com/stretchr/testify/suite"
+	"go.etcd.io/etcd/embed"
 )
 
 import (

registry/nacos/service_discovery_test.go

@@ -35,6 +35,7 @@ import (
 	"github.com/apache/dubbo-go/common/observer"
 	"github.com/apache/dubbo-go/common/observer/dispatcher"
 	"github.com/apache/dubbo-go/config"
+	"github.com/apache/dubbo-go/metadata/mapping"
 	"github.com/apache/dubbo-go/registry"
 	"github.com/apache/dubbo-go/registry/event"
 )
@@ -69,16 +70,6 @@ func Test_newNacosServiceDiscovery(t *testing.T) {
 	assert.NotNil(t, res)
 }
 
-func TestNacosServiceDiscovery_Destroy(t *testing.T) {
-	prepareData()
-	serviceDiscovery, err := extension.GetServiceDiscovery(constant.NACOS_KEY, testName)
-	assert.Nil(t, err)
-	assert.NotNil(t, serviceDiscovery)
-	err = serviceDiscovery.Destroy()
-	assert.Nil(t, err)
-	assert.Nil(t, serviceDiscovery.(*nacosServiceDiscovery).namingClient)
-}
-
 func TestNacosServiceDiscovery_CRUD(t *testing.T) {
 	if !checkNacosServerAlive() {
 		return
@@ -88,6 +79,10 @@ func TestNacosServiceDiscovery_CRUD(t *testing.T) {
 		return &dispatcher.MockEventDispatcher{}
 	})
 
+	extension.SetGlobalServiceNameMapping(func() mapping.ServiceNameMapping {
+		return &mockServiceNameMapping{}
+	})
+
 	extension.SetAndInitGlobalDispatcher("mock")
 	rand.Seed(time.Now().Unix())
 	serviceName := "service-name" + strconv.Itoa(rand.Intn(10000))
@@ -171,6 +166,16 @@ func TestNacosServiceDiscovery_GetDefaultPageSize(t *testing.T) {
 	assert.Equal(t, registry.DefaultPageSize, serviceDiscovery.GetDefaultPageSize())
 }
 
+func TestNacosServiceDiscovery_Destroy(t *testing.T) {
+	prepareData()
+	serviceDiscovery, err := extension.GetServiceDiscovery(constant.NACOS_KEY, testName)
+	assert.Nil(t, err)
+	assert.NotNil(t, serviceDiscovery)
+	err = serviceDiscovery.Destroy()
+	assert.Nil(t, err)
+	assert.Nil(t, serviceDiscovery.(*nacosServiceDiscovery).namingClient)
+}
+
 func prepareData() {
 	config.GetBaseConfig().ServiceDiscoveries[testName] = &config.ServiceDiscoveryConfig{
 		Protocol:  "nacos",
@@ -182,3 +187,13 @@ func prepareData() {
 		TimeoutStr: "10s",
 	}
 }
+
+type mockServiceNameMapping struct{}
+
+func (m *mockServiceNameMapping) Map(string, string, string, string) error {
+	return nil
+}
+
+func (m *mockServiceNameMapping) Get(string, string, string, string) (*gxset.HashSet, error) {
+	panic("implement me")
+}

registry/zookeeper/service_discovery_test.go

@@ -18,8 +18,7 @@
 package zookeeper
 
 import (
-	"github.com/apache/dubbo-go/registry/event"
-	gxset "github.com/dubbogo/gost/container/set"
+	"context"
 	"strconv"
 	"sync"
 	"testing"
@@ -27,14 +26,21 @@ import (
 
 import (
 	"github.com/dubbogo/go-zookeeper/zk"
+	gxset "github.com/dubbogo/gost/container/set"
 	"github.com/stretchr/testify/assert"
 )
 
 import (
+	"github.com/apache/dubbo-go/common"
+	"github.com/apache/dubbo-go/common/constant"
 	"github.com/apache/dubbo-go/common/extension"
 	"github.com/apache/dubbo-go/common/observer"
+	"github.com/apache/dubbo-go/common/observer/dispatcher"
 	"github.com/apache/dubbo-go/config"
+	"github.com/apache/dubbo-go/metadata/mapping"
+	"github.com/apache/dubbo-go/protocol"
 	"github.com/apache/dubbo-go/registry"
+	"github.com/apache/dubbo-go/registry/event"
 )
 
 var testName = "test"
@@ -80,22 +86,34 @@ func TestNewZookeeperServiceDiscovery(t *testing.T) {
 
 func TestCURDZookeeperServiceDiscovery(t *testing.T) {
 	prepareData(t)
+	extension.SetEventDispatcher("mock", func() observer.EventDispatcher {
+		return &dispatcher.MockEventDispatcher{}
+	})
+	extension.SetGlobalServiceNameMapping(func() mapping.ServiceNameMapping {
+		return &mockServiceNameMapping{}
+	})
+
+	extension.SetProtocol("mock", func() protocol.Protocol {
+		return &mockProtocol{}
+	})
+
 	sd, err := newZookeeperServiceDiscovery(testName)
 	assert.Nil(t, err)
 	defer func() {
 		_ = sd.Destroy()
 	}()
-	md := make(map[string]string)
-	md["t1"] = "test1"
-	err = sd.Register(&registry.DefaultServiceInstance{
+	ins := &registry.DefaultServiceInstance{
 		ID:          "testID",
 		ServiceName: testName,
 		Host:        "127.0.0.1",
 		Port:        2233,
 		Enable:      true,
 		Healthy:     true,
-		Metadata:    md,
-	})
+		Metadata:    nil,
+	}
+	ins.Metadata = map[string]string{"t1": "test1", constant.METADATA_SERVICE_URL_PARAMS_PROPERTY_NAME: `{"protocol":"mock","timeout":"10000","version":"1.0.0","dubbo":"2.0.2","release":"2.7.6","port":"2233"}`}
+	err = sd.Register(ins)
+
 	assert.Nil(t, err)
 
 	testsPager := sd.GetHealthyInstancesByPage(testName, 0, 1, true)
@@ -105,16 +123,18 @@ func TestCURDZookeeperServiceDiscovery(t *testing.T) {
 	assert.Equal(t, "127.0.0.1:2233", test.GetID())
 	assert.Equal(t, "test1", test.GetMetadata()["t1"])
 
-	md["t1"] = "test12"
-	err = sd.Update(&registry.DefaultServiceInstance{
+	ins = &registry.DefaultServiceInstance{
 		ID:          "testID",
 		ServiceName: testName,
 		Host:        "127.0.0.1",
 		Port:        2233,
 		Enable:      true,
 		Healthy:     true,
-		Metadata:    md,
-	})
+	}
+	ins.Metadata = map[string]string{"t1": "test12", constant.METADATA_SERVICE_URL_PARAMS_PROPERTY_NAME: `{"protocol":"mock","timeout":"10000","version":"1.0.0","dubbo":"2.0.2","release":"2.7.6","port":"2233"}`}
+
+	err = sd.Update(ins)
+
 	assert.Nil(t, err)
 
 	testsPager = sd.GetInstancesByPage(testName, 0, 1)
@@ -154,7 +174,7 @@ func TestAddListenerZookeeperServiceDiscovery(t *testing.T) {
 		_ = sd.Destroy()
 	}()
 
-	err = sd.Register(&registry.DefaultServiceInstance{
+	ins := &registry.DefaultServiceInstance{
 		ID:          "testID",
 		ServiceName: testName,
 		Host:        "127.0.0.1",
@@ -162,7 +182,10 @@ func TestAddListenerZookeeperServiceDiscovery(t *testing.T) {
 		Enable:      true,
 		Healthy:     true,
 		Metadata:    nil,
-	})
+	}
+	ins.Metadata = map[string]string{"t1": "test12", constant.METADATA_SERVICE_URL_PARAMS_PROPERTY_NAME: `{"protocol":"mock","timeout":"10000","version":"1.0.0","dubbo":"2.0.2","release":"2.7.6","port":"2233"}`}
+	err = sd.Register(ins)
+
 	assert.Nil(t, err)
 	wg := &sync.WaitGroup{}
 	wg.Add(1)
@@ -172,13 +195,15 @@ func TestAddListenerZookeeperServiceDiscovery(t *testing.T) {
 	}
 	hs := gxset.NewSet()
 	hs.Add(testName)
+
 	sicl := event.NewServiceInstancesChangedListener(hs)
+	sicl.AddListenerAndNotify(testName, tn)
 	extension.SetAndInitGlobalDispatcher("direct")
 	extension.GetGlobalDispatcher().AddEventListener(sicl)
 	err = sd.AddListener(sicl)
 	assert.NoError(t, err)
 
-	err = sd.Update(&registry.DefaultServiceInstance{
+	ins = &registry.DefaultServiceInstance{
 		ID:          "testID",
 		ServiceName: testName,
 		Host:        "127.0.0.1",
@@ -186,7 +211,9 @@ func TestAddListenerZookeeperServiceDiscovery(t *testing.T) {
 		Enable:      true,
 		Healthy:     true,
 		Metadata:    nil,
-	})
+	}
+	ins.Metadata = map[string]string{"t1": "test12", constant.METADATA_SERVICE_URL_PARAMS_PROPERTY_NAME: `{"protocol":"mock","timeout":"10000","version":"1.0.0","dubbo":"2.0.2","release":"2.7.6","port":"2233"}`}
+	err = sd.Update(ins)
 	assert.NoError(t, err)
 	tn.wg.Wait()
 }
@@ -196,9 +223,60 @@ type testNotify struct {
 	t  *testing.T
 }
 
-func (tn *testNotify) Notify(e observer.Event) {
-	ice := e.(*registry.ServiceInstancesChangedEvent)
-	assert.Equal(tn.t, 1, len(ice.Instances))
-	assert.Equal(tn.t, "127.0.0.1:2233", ice.Instances[0].GetID())
+func (tn *testNotify) Notify(e *registry.ServiceEvent) {
+	assert.Equal(tn.t, "2233", e.Service.Port)
 	tn.wg.Done()
 }
+func (tn *testNotify) NotifyAll([]*registry.ServiceEvent, func()) {
+
+}
+
+type mockServiceNameMapping struct{}
+
+func (m *mockServiceNameMapping) Map(string, string, string, string) error {
+	return nil
+}
+
+func (m *mockServiceNameMapping) Get(string, string, string, string) (*gxset.HashSet, error) {
+	return gxset.NewSet(config.GetApplicationConfig().Name), nil
+}
+
+type mockProtocol struct{}
+
+func (m mockProtocol) Export(protocol.Invoker) protocol.Exporter {
+	panic("implement me")
+}
+
+func (m mockProtocol) Refer(*common.URL) protocol.Invoker {
+	return &mockInvoker{}
+}
+
+func (m mockProtocol) Destroy() {
+	panic("implement me")
+}
+
+type mockInvoker struct{}
+
+func (m *mockInvoker) GetURL() *common.URL {
+	panic("implement me")
+}
+
+func (m *mockInvoker) IsAvailable() bool {
+	panic("implement me")
+}
+
+func (m *mockInvoker) Destroy() {
+	panic("implement me")
+}
+
+func (m *mockInvoker) Invoke(context.Context, protocol.Invocation) protocol.Result {
+	// for getMetadataInfo and ServiceInstancesChangedListenerImpl onEvent
+	serviceInfo := &common.ServiceInfo{ServiceKey: "test", MatchKey: "test"}
+	services := make(map[string]*common.ServiceInfo)
+	services["test"] = serviceInfo
+	return &protocol.RPCResult{
+		Rest: &common.MetadataInfo{
+			Services: services,
+		},
+	}
+}

remoting/etcdv3/listener.go

@@ -23,10 +23,10 @@ import (
 )
 
 import (
-	"github.com/coreos/etcd/clientv3"
-	"github.com/coreos/etcd/mvcc/mvccpb"
 	gxetcd "github.com/dubbogo/gost/database/kv/etcd/v3"
 	perrors "github.com/pkg/errors"
+	"go.etcd.io/etcd/clientv3"
+	"go.etcd.io/etcd/mvcc/mvccpb"
 )
 
 import (

remoting/etcdv3/listener_test.go

@@ -25,9 +25,9 @@ import (
 )
 
 import (
-	"github.com/coreos/etcd/embed"
 	gxetcd "github.com/dubbogo/gost/database/kv/etcd/v3"
 	"github.com/stretchr/testify/assert"
+	"go.etcd.io/etcd/embed"
 )
 
 import (
@@ -101,7 +101,7 @@ func TestListener(t *testing.T) {
 		{input: struct {
 			k string
 			v string
-		}{k: "/dubbo", v: changedData}},
+		}{k: "/dubbo/", v: changedData}},
 	}
 	SetUpEtcdServer(t)
 	c, err := gxetcd.NewClient("test", []string{"localhost:2381"}, time.Second, 1)
@@ -109,7 +109,7 @@ func TestListener(t *testing.T) {
 
 	listener := NewEventListener(c)
 	dataListener := &mockDataListener{client: c, changedData: changedData, rc: make(chan remoting.Event)}
-	listener.ListenServiceEvent("/dubbo", dataListener)
+	listener.ListenServiceEvent("/dubbo/", dataListener)
 
 	// NOTICE:  direct listen will lose create msg
 	time.Sleep(time.Second)
@@ -117,7 +117,7 @@ func TestListener(t *testing.T) {
 
 		k := tc.input.k
 		v := tc.input.v
-		if err := c.Create(k, v); err != nil {
+		if err := c.Update(k, v); err != nil {
 			t.Fatal(err)
 		}