Repository: apache/rocketmq-client-go
Pull request #708 (2021-08-07 01:46:35)

[ISSUE #735] centralized management errors
## What is the purpose of the change

Centralized management errors

## Brief changelog

move all errors.New() to errors.go



codecov-commenter: # [Codecov](https://codecov.io/gh/apache/rocketmq-client-go/pull/708?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#708](https://codecov.io/gh/apache/rocketmq-client-go/pull/708?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (98eaad3) into [master](https://codecov.io/gh/apache/rocketmq-client-go/commit/46701f1d6e4b3fbbbd1e702a0cae9ae947b91da3?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (46701f1) will **not change** coverage.
> The diff coverage is `29.41%`.

[![Impacted file tree graph](https://codecov.io/gh/apache/rocketmq-client-go/pull/708/graphs/tree.svg?width=650&height=150&src=pr&token=CiV9ofPkG3&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)](https://codecov.io/gh/apache/rocketmq-client-go/pull/708?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)

```diff
@@           Coverage Diff           @@
##           master     #708   +/-   ##
=======================================
  Coverage   25.70%   25.70%           
=======================================
  Files          49       49           
  Lines        6123     6123           
=======================================
  Hits         1574     1574           
  Misses       4335     4335           
  Partials      214      214           
```


| [Impacted Files](https://codecov.io/gh/apache/rocketmq-client-go/pull/708?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [consumer/consumer.go](https://codecov.io/gh/apache/rocketmq-client-go/pull/708/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Y29uc3VtZXIvY29uc3VtZXIuZ28=) | `16.34% <0.00%> (ø)` | |
| [consumer/pull\_consumer.go](https://codecov.io/gh/apache/rocketmq-client-go/pull/708/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Y29uc3VtZXIvcHVsbF9jb25zdW1lci5nbw==) | `0.00% <0.00%> (ø)` | |
| [consumer/push\_consumer.go](https://codecov.io/gh/apache/rocketmq-client-go/pull/708/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Y29uc3VtZXIvcHVzaF9jb25zdW1lci5nbw==) | `12.16% <0.00%> (ø)` | |
| [internal/client.go](https://codecov.io/gh/apache/rocketmq-client-go/pull/708/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-aW50ZXJuYWwvY2xpZW50Lmdv) | `0.19% <ø> (ø)` | |
| [internal/utils/compression.go](https://codecov.io/gh/apache/rocketmq-client-go/pull/708/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-aW50ZXJuYWwvdXRpbHMvY29tcHJlc3Npb24uZ28=) | `75.00% <0.00%> (ø)` | |
| [internal/utils/errors.go](https://codecov.io/gh/apache/rocketmq-client-go/pull/708/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-aW50ZXJuYWwvdXRpbHMvZXJyb3JzLmdv) | `0.00% <ø> (ø)` | |
| [internal/utils/net.go](https://codecov.io/gh/apache/rocketmq-client-go/pull/708/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-aW50ZXJuYWwvdXRpbHMvbmV0Lmdv) | `56.52% <0.00%> (ø)` | |
| [primitive/errors.go](https://codecov.io/gh/apache/rocketmq-client-go/pull/708/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cHJpbWl0aXZlL2Vycm9ycy5nbw==) | `0.00% <ø> (ø)` | |
| [producer/producer.go](https://codecov.io/gh/apache/rocketmq-client-go/pull/708/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cHJvZHVjZXIvcHJvZHVjZXIuZ28=) | `30.59% <33.33%> (ø)` | |
| [primitive/base.go](https://codecov.io/gh/apache/rocketmq-client-go/pull/708/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cHJpbWl0aXZlL2Jhc2UuZ28=) | `41.17% <50.00%> (ø)` | |
| ... and [2 more](https://codecov.io/gh/apache/rocketmq-client-go/pull/708/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/apache/rocketmq-client-go/pull/708?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/rocketmq-client-go/pull/708?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Last update [46701f1...98eaad3](https://codecov.io/gh/apache/rocketmq-client-go/pull/708?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).


Patch:
diff --git a/.gitignore b/.gitignore
index 99fd04d..6644c6b 100644
--- a/.gitignore
+++ b/.gitignore
@@ -1,6 +1,4 @@
 .idea
-go.mod
-go.sum
 vendor/
 coverage.txt
 examples/test
diff --git a/.travis.yml b/.travis.yml
index d4ed1ac..5073ebc 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -17,6 +17,7 @@ env:
     - OS_TYPE=centos OS_VERSION=7
 
 before_script:
+  - export PROJECT_PATH=`pwd`
   - cd ${TRAVIS_HOME}
   - wget https://archive.apache.org/dist/rocketmq/4.6.0/rocketmq-all-4.6.0-bin-release.zip
   - unzip rocketmq-all-4.6.0-bin-release.zip
@@ -26,12 +27,9 @@ before_script:
   - nohup sh bin/mqbroker -n localhost:9876 &
 
 script:
-  - cd ${TRAVIS_HOME}
-  - ls -al
-  - cd ${GOPATH}/src/github.com/apache/rocketmq-client-go/v2
-  - ls -al
+  - cd ${PROJECT_PATH}
   - go fmt ./... && [[ -z `git status -s` ]]
-  - go mod vendor && go test ./... -coverprofile=coverage.txt -covermode=atomic
+  - go test ./... -coverprofile=coverage.txt -covermode=atomic
 
 after_success:
-  - bash <(curl -s https://codecov.io/bash)
+  - bash < (curl -s https://codecov.io/bash)
diff --git a/NOTICE b/NOTICE
index 65ebdd0..086ee9f 100644
--- a/NOTICE
+++ b/NOTICE
@@ -1,5 +1,5 @@
 Apache RocketMQ
-Copyright 2016-2020 The Apache Software Foundation
+Copyright 2016-2021 The Apache Software Foundation
 
 This product includes software developed at
 The Apache Software Foundation (http://www.apache.org/).
diff --git a/api.go b/api.go
index 0e149e9..31f58d5 100644
--- a/api.go
+++ b/api.go
@@ -19,8 +19,7 @@ package rocketmq
 
 import (
 	"context"
-
-	"github.com/pkg/errors"
+	"github.com/apache/rocketmq-client-go/v2/errors"
 
 	"github.com/apache/rocketmq-client-go/v2/consumer"
 	"github.com/apache/rocketmq-client-go/v2/primitive"
@@ -134,5 +133,5 @@ type PullConsumer interface {
 //
 // The PullConsumer will be supported in next release
 func NewPullConsumer(opts ...consumer.Option) (PullConsumer, error) {
-	return nil, errors.New("pull consumer has not supported")
+	return nil, errors.ErrPullConsumer
 }
diff --git a/benchmark/stable.go b/benchmark/stable.go
index 2c3dc1b..cd5fb9b 100644
--- a/benchmark/stable.go
+++ b/benchmark/stable.go
@@ -18,9 +18,9 @@
 package main
 
 import (
-	"errors"
 	"flag"
 	"fmt"
+	"github.com/apache/rocketmq-client-go/v2/errors"
 	"os"
 	"os/signal"
 	"syscall"
@@ -52,23 +52,23 @@ func (st *stableTest) buildFlags(name string) {
 
 func (st *stableTest) checkFlag() error {
 	if st.topic == "" {
-		return errors.New("empty topic")
+		return errors.ErrEmptyTopic
 	}
 
 	if st.nameSrv == "" {
-		return errors.New("empty namesrv")
+		return errors.ErrEmptyNameSrv
 	}
 
 	if st.groupID == "" {
-		return errors.New("empty group id")
+		return errors.ErrEmptyGroupID
 	}
 
 	if st.testMin <= 0 {
-		return errors.New("test miniutes must be positive integer")
+		return errors.ErrTestMin
 	}
 
 	if st.opIntervalSec <= 0 {
-		return errors.New("operation interval must be positive integer")
+		return errors.ErrOperationInterval
 	}
 
 	return nil
@@ -114,7 +114,7 @@ func (stp *stableTestProducer) checkFlag() error {
 		return err
 	}
 	if stp.bodySize <= 0 {
-		return errors.New("message body size must be positive integer")
+		return errors.ErrMessageBody
 	}
 
 	return nil
@@ -187,7 +187,7 @@ func (stc *stableTestConsumer) checkFlag() error {
 	}
 
 	if stc.expression == "" {
-		return errors.New("empty expression")
+		return errors.ErrEmptyExpression
 	}
 	return nil
 }
diff --git a/consumer/consumer.go b/consumer/consumer.go
index 42bc973..ad77bc1 100644
--- a/consumer/consumer.go
+++ b/consumer/consumer.go
@@ -20,6 +20,7 @@ package consumer
 import (
 	"context"
 	"fmt"
+	"github.com/apache/rocketmq-client-go/v2/errors"
 	"sort"
 	"strconv"
 	"strings"
@@ -29,7 +30,6 @@ import (
 
 	jsoniter "github.com/json-iterator/go"
 
-	"github.com/pkg/errors"
 	"github.com/tidwall/gjson"
 
 	"github.com/apache/rocketmq-client-go/v2/internal"
@@ -68,11 +68,6 @@ const (
 	_SubAll = "*"
 )
 
-var (
-	ErrCreated        = errors.New("consumer group has been created")
-	ErrBrokerNotFound = errors.New("broker can not found")
-)
-
 // Message model defines the way how messages are delivered to each consumer clients.
 // </p>
 //
@@ -822,7 +817,7 @@ func (dc *defaultConsumer) pullInner(ctx context.Context, queue *primitive.Messa
 		rlog.Warning("no broker found for mq", map[string]interface{}{
 			rlog.LogKeyMessageQueue: queue,
 		})
-		return nil, ErrBrokerNotFound
+		return nil, errors.ErrBrokerNotFound
 	}
 
 	if brokerResult.Slave {
@@ -993,6 +988,7 @@ func buildSubscriptionData(topic string, selector MessageSelector) *internal.Sub
 		SubString: selector.Expression,
 		ExpType:   string(selector.Type),
 	}
+	subData.SubVersion = time.Now().UnixNano()
 
 	if selector.Type != "" && selector.Type != TAG {
 		return subData
diff --git a/consumer/pull_consumer.go b/consumer/pull_consumer.go
index 0b0ef56..81a3ec5 100644
--- a/consumer/pull_consumer.go
+++ b/consumer/pull_consumer.go
@@ -20,13 +20,13 @@ package consumer
 import (
 	"context"
 	"fmt"
+	errors2 "github.com/apache/rocketmq-client-go/v2/errors"
 	"sync"
 	"sync/atomic"
 
 	"github.com/pkg/errors"
 
 	"github.com/apache/rocketmq-client-go/v2/internal"
-	"github.com/apache/rocketmq-client-go/v2/internal/utils"
 	"github.com/apache/rocketmq-client-go/v2/primitive"
 	"github.com/apache/rocketmq-client-go/v2/rlog"
 )
@@ -93,7 +93,7 @@ func NewPullConsumer(options ...Option) (*defaultPullConsumer, error) {
 
 		namesrv: srvs,
 	}
-
+	dc.option.ClientOptions.Namesrv, err = internal.GetNamesrv(dc.client.ClientID())
 	c := &defaultPullConsumer{
 		defaultConsumer: dc,
 	}
@@ -174,15 +174,15 @@ func (dc *defaultConsumer) checkPull(ctx context.Context, mq *primitive.MessageQ
 	}
 
 	if mq == nil {
-		return utils.ErrMQEmpty
+		return errors2.ErrMQEmpty
 	}
 
 	if offset < 0 {
-		return utils.ErrOffset
+		return errors2.ErrOffset
 	}
 
 	if numbers <= 0 {
-		return utils.ErrNumbers
+		return errors2.ErrNumbers
 	}
 	return nil
 }
diff --git a/consumer/push_consumer.go b/consumer/push_consumer.go
index c84ce84..8827a87 100644
--- a/consumer/push_consumer.go
+++ b/consumer/push_consumer.go
@@ -20,6 +20,7 @@ package consumer
 import (
 	"context"
 	"fmt"
+	errors2 "github.com/apache/rocketmq-client-go/v2/errors"
 	"math"
 	"strconv"
 	"strings"
@@ -99,10 +100,14 @@ func NewPushConsumer(opts ...Option) (*pushConsumer, error) {
 		consumeOrderly: defaultOpts.ConsumeOrderly,
 		fromWhere:      defaultOpts.FromWhere,
 		allocate:       defaultOpts.Strategy,
-		option:         defaultOpts,
 		namesrv:        srvs,
+		option:         defaultOpts,
 	}
-
+	dc.option.ClientOptions.Namesrv, err = internal.GetNamesrv(dc.client.ClientID())
+	if err != nil {
+		return nil, err
+	}
+	dc.namesrv = dc.option.ClientOptions.Namesrv
 	p := &pushConsumer{
 		defaultConsumer: dc,
 		subscribedTopic: make(map[string]string, 0),
@@ -119,6 +124,11 @@ func NewPushConsumer(opts ...Option) (*pushConsumer, error) {
 
 	p.interceptor = primitive.ChainInterceptors(p.option.Interceptors...)
 
+	if p.model == Clustering {
+		retryTopic := internal.GetRetryTopic(p.consumerGroup)
+		sub := buildSubscriptionData(retryTopic, MessageSelector{TAG, _SubAll})
+		p.subscriptionDataTable.Store(retryTopic, sub)
+	}
 	return p, nil
 }
 
@@ -138,7 +148,7 @@ func (pc *pushConsumer) Start() error {
 			rlog.Error("the consumer group has been created, specify another one", map[string]interface{}{
 				rlog.LogKeyConsumerGroup: pc.consumerGroup,
 			})
-			err = ErrCreated
+			err = errors2.ErrCreated
 			return
 		}
 
@@ -224,15 +234,7 @@ func (pc *pushConsumer) Subscribe(topic string, selector MessageSelector,
 	f func(context.Context, ...*primitive.MessageExt) (ConsumeResult, error)) error {
 	if atomic.LoadInt32(&pc.state) == int32(internal.StateStartFailed) ||
 		atomic.LoadInt32(&pc.state) == int32(internal.StateShutdown) {
-		return errors.New("cannot subscribe topic since client either failed to start or has been shutdown.")
-	}
-
-	// add retry topic subscription for resubscribe
-	retryTopic := internal.GetRetryTopic(pc.consumerGroup)
-	_, exists := pc.subscriptionDataTable.Load(retryTopic)
-	if !exists {
-		sub := buildSubscriptionData(retryTopic, MessageSelector{TAG, _SubAll})
-		pc.subscriptionDataTable.Store(retryTopic, sub)
+		return errors2.ErrStartTopic
 	}
 
 	if pc.option.Namespace != "" {
@@ -254,8 +256,6 @@ func (pc *pushConsumer) Unsubscribe(topic string) error {
 		topic = pc.option.Namespace + "%" + topic
 	}
 	pc.subscriptionDataTable.Delete(topic)
-	retryTopic := internal.GetRetryTopic(pc.consumerGroup)
-	pc.subscriptionDataTable.Delete(retryTopic)
 	return nil
 }
 
diff --git a/consumer/strategy.go b/consumer/strategy.go
index 2af48de..4a07928 100644
--- a/consumer/strategy.go
+++ b/consumer/strategy.go
@@ -20,7 +20,7 @@ package consumer
 import (
 	"strings"
 
-	"stathat.com/c/consistent"
+	"github.com/stathat/consistent"
 
 	"github.com/apache/rocketmq-client-go/v2/internal/utils"
 	"github.com/apache/rocketmq-client-go/v2/primitive"
diff --git a/errors.go b/errors.go
deleted file mode 100644
index fe9ba33..0000000
--- a/errors.go
+++ /dev/null
@@ -1,29 +0,0 @@
-/*
-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 rocketmq
-
-import (
-	"github.com/pkg/errors"
-)
-
-var (
-	ErrRequestTimeout = errors.New("request timeout")
-	ErrMQEmpty        = errors.New("MessageQueue is nil")
-	ErrOffset         = errors.New("offset < 0")
-	ErrNumbers        = errors.New("numbers < 0")
-)
diff --git a/errors/errors.go b/errors/errors.go
new file mode 100644
index 0000000..793fcda
--- /dev/null
+++ b/errors/errors.go
@@ -0,0 +1,50 @@
+/*
+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 errors
+
+import "errors"
+
+var (
+	ErrRequestTimeout    = errors.New("equest timeout")
+	ErrMQEmpty           = errors.New("MessageQueue is nil")
+	ErrOffset            = errors.New("offset < 0")
+	ErrNumbers           = errors.New("numbers < 0")
+	ErrEmptyTopic        = errors.New("empty topic")
+	ErrEmptyNameSrv      = errors.New("empty namesrv")
+	ErrEmptyGroupID      = errors.New("empty group id")
+	ErrTestMin           = errors.New("test minutes must be positive integer")
+	ErrOperationInterval = errors.New("operation interval must be positive integer")
+	ErrMessageBody       = errors.New("message body size must be positive integer")
+	ErrEmptyExpression   = errors.New("empty expression")
+	ErrCreated           = errors.New("consumer group has been created")
+	ErrBrokerNotFound    = errors.New("broker can not found")
+	ErrStartTopic        = errors.New("cannot subscribe topic since client either failed to start or has been shutdown.")
+	ErrResponse          = errors.New("response error")
+	ErrCompressLevel     = errors.New("unsupported compress level")
+	ErrUnknownIP         = errors.New("unknown IP address")
+	ErrService           = errors.New("service close is not running, please check")
+	ErrTopicNotExist     = errors.New("topic not exist")
+	ErrNotExisted        = errors.New("not existed")
+	ErrNoNameserver      = errors.New("nameServerAddrs can't be empty.")
+	ErrMultiIP           = errors.New("multiple IP addr does not support")
+	ErrIllegalIP         = errors.New("IP addr error")
+	ErrTopicEmpty        = errors.New("topic is nil")
+	ErrMessageEmpty      = errors.New("message is nil")
+	ErrNotRunning        = errors.New("producer not started")
+	ErrPullConsumer      = errors.New("pull consumer has not supported")
+)
diff --git a/examples/consumer/pull/main.go b/examples/consumer/pull/main.go
index 3d49e7e..5b5819e 100644
--- a/examples/consumer/pull/main.go
+++ b/examples/consumer/pull/main.go
@@ -20,6 +20,7 @@ package main
 import (
 	"context"
 	"fmt"
+	"github.com/apache/rocketmq-client-go/v2/errors"
 	"time"
 
 	"github.com/apache/rocketmq-client-go/v2"
@@ -52,7 +53,7 @@ func main() {
 	for {
 		resp, err := c.PullFrom(ctx, queue, offset, 10)
 		if err != nil {
-			if err == rocketmq.ErrRequestTimeout {
+			if err == errors.ErrRequestTimeout {
 				fmt.Printf("timeout \n")
 				time.Sleep(1 * time.Second)
 				continue
diff --git a/go.mod b/go.mod
index 8388959..80de649 100644
--- a/go.mod
+++ b/go.mod
@@ -1,20 +1,17 @@
 module github.com/apache/rocketmq-client-go/v2
 
+go 1.13
+
 require (
 	github.com/emirpasic/gods v1.12.0
 	github.com/golang/mock v1.3.1
 	github.com/json-iterator/go v1.1.9
 	github.com/pkg/errors v0.8.1
 	github.com/sirupsen/logrus v1.4.1
-	github.com/smartystreets/goconvey v0.0.0-20190710185942-9d28bd7c0945
+	github.com/smartystreets/goconvey v1.6.4
+	github.com/stathat/consistent v1.0.0
 	github.com/stretchr/testify v1.3.0
-	github.com/tidwall/gjson v1.2.1
-	github.com/tidwall/match v1.0.1 // indirect
-	github.com/tidwall/pretty v0.0.0-20190325153808-1166b9ac2b65 // indirect
+	github.com/tidwall/gjson v1.8.1
 	go.uber.org/atomic v1.5.1
-	stathat.com/c/consistent v1.0.0
+	stathat.com/c/consistent v1.0.0 // indirect
 )
-
-replace stathat.com/c/consistent v1.0.0 => github.com/stathat/consistent v1.0.0
-
-go 1.13
diff --git a/go.sum b/go.sum
index 3b724db..58bed76 100644
--- a/go.sum
+++ b/go.sum
@@ -26,8 +26,8 @@ github.com/sirupsen/logrus v1.4.1 h1:GL2rEmy6nsikmW0r8opw9JIRScdMF5hA8cOYLH7In1k
 github.com/sirupsen/logrus v1.4.1/go.mod h1:ni0Sbl8bgC9z8RoU9G6nDWqqs/fq4eDPysMBDgk/93Q=
 github.com/smartystreets/assertions v0.0.0-20180927180507-b2de0cb4f26d h1:zE9ykElWQ6/NYmHa3jpm/yHnI4xSofP+UP6SpjHcSeM=
 github.com/smartystreets/assertions v0.0.0-20180927180507-b2de0cb4f26d/go.mod h1:OnSkiWE9lh6wB0YB77sQom3nweQdgAjqCqsofrRNTgc=
-github.com/smartystreets/goconvey v0.0.0-20190710185942-9d28bd7c0945 h1:N8Bg45zpk/UcpNGnfJt2y/3lRWASHNTUET8owPYCgYI=
-github.com/smartystreets/goconvey v0.0.0-20190710185942-9d28bd7c0945/go.mod h1:syvi0/a8iFYH4r/RixwvyeAJjdLS9QV7WQ/tjFTllLA=
+github.com/smartystreets/goconvey v1.6.4 h1:fv0U8FUIMPNf1L9lnHLvLhgicrIVChEkdzIKYqbNC9s=
+github.com/smartystreets/goconvey v1.6.4/go.mod h1:syvi0/a8iFYH4r/RixwvyeAJjdLS9QV7WQ/tjFTllLA=
 github.com/stathat/consistent v1.0.0 h1:ZFJ1QTRn8npNBKW065raSZ8xfOqhpb8vLOkfp4CcL/U=
 github.com/stathat/consistent v1.0.0/go.mod h1:uajTPbgSygZBJ+V+0mY7meZ8i0XAcZs7AQ6V121XSxw=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
@@ -35,12 +35,12 @@ github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+
 github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
 github.com/stretchr/testify v1.3.0 h1:TivCn/peBQ7UY8ooIcPgZFpTNSz0Q2U6UrFlUfqbe0Q=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
-github.com/tidwall/gjson v1.2.1 h1:j0efZLrZUvNerEf6xqoi0NjWMK5YlLrR7Guo/dxY174=
-github.com/tidwall/gjson v1.2.1/go.mod h1:c/nTNbUr0E0OrXEhq1pwa8iEgc2DOt4ZZqAt1HtCkPA=
-github.com/tidwall/match v1.0.1 h1:PnKP62LPNxHKTwvHHZZzdOAOCtsJTjo6dZLCwpKm5xc=
-github.com/tidwall/match v1.0.1/go.mod h1:LujAq0jyVjBy028G1WhWfIzbpQfMO8bBZ6Tyb0+pL9E=
-github.com/tidwall/pretty v0.0.0-20190325153808-1166b9ac2b65 h1:rQ229MBgvW68s1/g6f1/63TgYwYxfF4E+bi/KC19P8g=
-github.com/tidwall/pretty v0.0.0-20190325153808-1166b9ac2b65/go.mod h1:XNkn88O1ChpSDQmQeStsy+sBenx6DDtFZJxhVysOjyk=
+github.com/tidwall/gjson v1.8.1 h1:8j5EE9Hrh3l9Od1OIEDAb7IpezNA20UdRngNAj5N0WU=
+github.com/tidwall/gjson v1.8.1/go.mod h1:5/xDoumyyDNerp2U36lyolv46b3uF/9Bu6OfyQ9GImk=
+github.com/tidwall/match v1.0.3 h1:FQUVvBImDutD8wJLN6c5eMzWtjgONK9MwIBCOrUJKeE=
+github.com/tidwall/match v1.0.3/go.mod h1:eRSPERbgtNPcGhD8UCthc6PmLEQXEWd3PRB5JTxsfmM=
+github.com/tidwall/pretty v1.1.0 h1:K3hMW5epkdAVwibsQEfR/7Zj0Qgt4DxtNumTq/VloO8=
+github.com/tidwall/pretty v1.1.0/go.mod h1:XNkn88O1ChpSDQmQeStsy+sBenx6DDtFZJxhVysOjyk=
 go.uber.org/atomic v1.5.1 h1:rsqfU5vBkVknbhUGbAUwQKR2H4ItV8tjJ+6kJX4cxHM=
 go.uber.org/atomic v1.5.1/go.mod h1:sABNBOSYdrvTF6hTgEIbc7YasKWGhgEQZyfxyTvoXHQ=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
@@ -49,15 +49,15 @@ golang.org/x/lint v0.0.0-20190930215403-16217165b5de/go.mod h1:6SW0HCj/g11FgYtHl
 golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
-golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33 h1:I6FyU15t786LL7oL/hn43zqTuEGr4PN7F4XJ1p4E3Y8=
 golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a h1:1BGLXjeY4akVXGgbC9HugT3Jv3hCI0z56oJR5vAMgBU=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/tools v0.0.0-20190311212946-11955173bddd/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
 golang.org/x/tools v0.0.0-20190328211700-ab21143f2384/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
-golang.org/x/tools v0.0.0-20190425150028-36563e24a262 h1:qsl9y/CJx34tuA7QCPNp86JNJe4spst6Ff8MjvPUdPg=
 golang.org/x/tools v0.0.0-20190425150028-36563e24a262/go.mod h1:RgjU9mgBXZiqYHBnxXauZ1Gv1EHHAz9KjViQ78xBX0Q=
 golang.org/x/tools v0.0.0-20191029041327-9cc4af7d6b2c h1:IGkKhmfzcztjm6gYkykvu/NiS8kaqbCWAEWWAyf8J5U=
 golang.org/x/tools v0.0.0-20191029041327-9cc4af7d6b2c/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+stathat.com/c/consistent v1.0.0 h1:ezyc51EGcRPJUxfHGSgJjWzJdj3NiMU9pNfLNGiXV0c=
+stathat.com/c/consistent v1.0.0/go.mod h1:QkzMWzcbB+yQBL2AttO6sgsQS/JSTapcDISJalmCDS0=
diff --git a/internal/client.go b/internal/client.go
index 3a09ea8..35a88a4 100644
--- a/internal/client.go
+++ b/internal/client.go
@@ -21,6 +21,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	errors2 "github.com/apache/rocketmq-client-go/v2/errors"
 	"net"
 	"os"
 	"strconv"
@@ -55,7 +56,7 @@ const (
 )
 
 var (
-	ErrServiceState = errors.New("service close is not running, please check")
+	ErrServiceState = errors2.ErrService
 
 	_VIPChannelEnable = false
 )
@@ -186,6 +187,9 @@ func GetOrNewRocketMQClient(option ClientOptions, callbackCh chan interface{}) R
 		done:         make(chan struct{}),
 	}
 	actual, loaded := clientMap.LoadOrStore(client.ClientID(), client)
+	client.namesrvs = GetOrSetNamesrv(client.ClientID(), client.namesrvs)
+	client.namesrvs.bundleClient = actual.(*rmqClient)
+	client.option.Namesrv = client.namesrvs
 	if !loaded {
 		client.remoteClient.RegisterRequestFunc(ReqNotifyConsumerIdsChanged, func(req *remote.RemotingCommand, addr net.Addr) *remote.RemotingCommand {
 			rlog.Info("receive broker's notification to consumer group", map[string]interface{}{
diff --git a/internal/namesrv.go b/internal/namesrv.go
index a47bbc1..7776651 100644
--- a/internal/namesrv.go
+++ b/internal/namesrv.go
@@ -19,6 +19,7 @@ package internal
 
 import (
 	"errors"
+	"fmt"
 	"regexp"
 	"strings"
 	"sync"
@@ -76,6 +77,8 @@ type namesrvs struct {
 	// brokerName -> *BrokerData
 	brokerAddressesMap sync.Map
 
+	bundleClient *rmqClient
+
 	// brokerName -> map[string]int32: brokerAddr -> version
 	brokerVersionMap map[string]map[string]int32
 	// lock for broker version read/write
@@ -92,9 +95,21 @@ type namesrvs struct {
 }
 
 var _ Namesrvs = (*namesrvs)(nil)
+var namesrvMap sync.Map
 
 // NewNamesrv init Namesrv from namesrv addr string.
 // addr primitive.NamesrvAddr
+func GetOrSetNamesrv(clientId string, namesrv *namesrvs) *namesrvs {
+	actual, _ := namesrvMap.LoadOrStore(clientId, namesrv)
+	return actual.(*namesrvs)
+}
+func GetNamesrv(clientId string) (*namesrvs, error) {
+	actual, ok := namesrvMap.Load(clientId)
+	if !ok {
+		return nil, fmt.Errorf("the namesrv in instanceName [%s] not found", clientId)
+	}
+	return actual.(*namesrvs), nil
+}
 func NewNamesrv(resolver primitive.NsResolver) (*namesrvs, error) {
 	addr := resolver.Resolve()
 	if len(addr) == 0 {
diff --git a/internal/remote/future.go b/internal/remote/future.go
index ffbf781..a7d3268 100644
--- a/internal/remote/future.go
+++ b/internal/remote/future.go
@@ -19,9 +19,8 @@ package remote
 
 import (
 	"context"
+	"github.com/apache/rocketmq-client-go/v2/errors"
 	"sync"
-
-	"github.com/apache/rocketmq-client-go/v2/internal/utils"
 )
 
 // ResponseFuture
@@ -62,7 +61,7 @@ func (r *ResponseFuture) waitResponse() (*RemotingCommand, error) {
 	case <-r.Done:
 		cmd, err = r.ResponseCommand, r.Err
 	case <-r.ctx.Done():
-		err = utils.ErrRequestTimeout
+		err = errors.ErrRequestTimeout
 		r.Err = err
 	}
 	return cmd, err
diff --git a/internal/route.go b/internal/route.go
index 66be96d..9cfa398 100644
--- a/internal/route.go
+++ b/internal/route.go
@@ -19,7 +19,7 @@ package internal
 
 import (
 	"context"
-	"errors"
+	"github.com/apache/rocketmq-client-go/v2/errors"
 	"math/rand"
 	"sort"
 	"strconv"
@@ -46,10 +46,6 @@ const (
 	MasterId         = int64(0)
 )
 
-var (
-	ErrTopicNotExist = errors.New("topic not exist")
-)
-
 func (s *namesrvs) cleanOfflineBroker() {
 	// TODO optimize
 	s.lockNamesrv.Lock()
@@ -161,6 +157,35 @@ func (s *namesrvs) UpdateTopicRouteInfoWithDefault(topic string, defaultTopic st
 	}
 
 	if changed {
+		if s.bundleClient != nil {
+			s.bundleClient.producerMap.Range(func(key, value interface{}) bool {
+				p := value.(InnerProducer)
+				updated := changed
+				if !updated {
+					updated = p.IsPublishTopicNeedUpdate(topic)
+				}
+				if updated {
+					publishInfo := s.bundleClient.namesrvs.routeData2PublishInfo(topic, routeData)
+					publishInfo.HaveTopicRouterInfo = true
+					p.UpdateTopicPublishInfo(topic, publishInfo)
+				}
+				return true
+			})
+			s.bundleClient.consumerMap.Range(func(key, value interface{}) bool {
+				consumer := value.(InnerConsumer)
+				updated := changed
+				if !updated {
+					updated = consumer.IsSubscribeTopicNeedUpdate(topic)
+				}
+				if updated {
+					consumer.UpdateTopicSubscribeInfo(topic, routeData2SubscribeInfo(topic, routeData))
+				}
+
+				return true
+			})
+			rlog.Info("change the route for clients", nil)
+		}
+
 		s.routeDataMap.Store(topic, routeData)
 		rlog.Info("the topic route info changed", map[string]interface{}{
 			rlog.LogKeyTopic:            topic,
@@ -405,7 +430,7 @@ func (s *namesrvs) queryTopicRouteInfoFromServer(topic string) (*TopicRouteData,
 		}
 		return routeData, nil
 	case ResTopicNotExist:
-		return nil, ErrTopicNotExist
+		return nil, errors.ErrTopicNotExist
 	default:
 		return nil, primitive.NewMQClientErr(response.Code, response.Remark)
 	}
diff --git a/internal/utils/compression.go b/internal/utils/compression.go
index 379cdf2..162864f 100644
--- a/internal/utils/compression.go
+++ b/internal/utils/compression.go
@@ -20,7 +20,7 @@ package utils
 import (
 	"bytes"
 	"compress/zlib"
-	"errors"
+	"github.com/apache/rocketmq-client-go/v2/errors"
 	"io/ioutil"
 	"sync"
 )
@@ -48,7 +48,7 @@ func init() {
 
 func Compress(raw []byte, compressLevel int) ([]byte, error) {
 	if compressLevel < zlib.BestSpeed || compressLevel > zlib.BestCompression {
-		return nil, errors.New("unsupported compress level")
+		return nil, errors.ErrCompressLevel
 	}
 
 	buf := bufPool.Get().(*bytes.Buffer)
diff --git a/internal/utils/errors.go b/internal/utils/errors.go
index 0b7ffc2..0887a37 100644
--- a/internal/utils/errors.go
+++ b/internal/utils/errors.go
@@ -18,19 +18,9 @@ limitations under the License.
 package utils
 
 import (
-	"errors"
-
 	"github.com/apache/rocketmq-client-go/v2/rlog"
 )
 
-var (
-	// ErrRequestTimeout for request timeout error
-	ErrRequestTimeout = errors.New("request timeout")
-	ErrMQEmpty        = errors.New("MessageQueue is nil")
-	ErrOffset         = errors.New("offset < 0")
-	ErrNumbers        = errors.New("numbers < 0")
-)
-
 func CheckError(action string, err error) {
 	if err != nil {
 		rlog.Error(action, map[string]interface{}{
diff --git a/internal/utils/net.go b/internal/utils/net.go
index cf35594..a4eeb56 100644
--- a/internal/utils/net.go
+++ b/internal/utils/net.go
@@ -19,8 +19,8 @@ package utils
 
 import (
 	"bytes"
-	"errors"
 	"fmt"
+	"github.com/apache/rocketmq-client-go/v2/errors"
 	"net"
 	"strconv"
 	"time"
@@ -56,7 +56,7 @@ func ClientIP4() ([]byte, error) {
 			}
 		}
 	}
-	return nil, errors.New("unknown IP address")
+	return nil, errors.ErrUnknownIP
 }
 
 func FakeIP() []byte {
diff --git a/primitive/base.go b/primitive/base.go
index fca34a2..a45fbb9 100644
--- a/primitive/base.go
+++ b/primitive/base.go
@@ -18,6 +18,7 @@ limitations under the License.
 package primitive
 
 import (
+	"github.com/apache/rocketmq-client-go/v2/errors"
 	"regexp"
 	"strings"
 )
@@ -31,7 +32,7 @@ type NamesrvAddr []string
 
 func NewNamesrvAddr(s ...string) (NamesrvAddr, error) {
 	if len(s) == 0 {
-		return nil, ErrNoNameserver
+		return nil, errors.ErrNoNameserver
 	}
 
 	ss := s
@@ -69,17 +70,17 @@ func verifyIP(ip string) error {
 		return nil
 	}
 	if strings.Contains(ip, ";") {
-		return ErrMultiIP
+		return errors.ErrMultiIP
 	}
 	ipV4s := ipv4Regex.FindAllString(ip, -1)
 	ipV6s := ipv6Regex.FindAllString(ip, -1)
 
 	if len(ipV4s) == 0 && len(ipV6s) == 0 {
-		return ErrIllegalIP
+		return errors.ErrIllegalIP
 	}
 
 	if len(ipV4s) > 1 || len(ipV6s) > 1 {
-		return ErrMultiIP
+		return errors.ErrMultiIP
 	}
 	return nil
 }
diff --git a/primitive/errors.go b/primitive/errors.go
index 9b1a88e..0fc8a7f 100644
--- a/primitive/errors.go
+++ b/primitive/errors.go
@@ -18,16 +18,9 @@ limitations under the License.
 package primitive
 
 import (
-	"errors"
 	"strconv"
 )
 
-var (
-	ErrNoNameserver = errors.New("nameServerAddrs can't be empty.")
-	ErrMultiIP      = errors.New("multiple IP addr does not support")
-	ErrIllegalIP    = errors.New("IP addr error")
-)
-
 type MQBrokerErr struct {
 	ResponseCode int16
 	ErrorMessage string
diff --git a/producer/interceptor.go b/producer/interceptor.go
index b3d2598..160deac 100644
--- a/producer/interceptor.go
+++ b/producer/interceptor.go
@@ -50,7 +50,7 @@ func newTraceInterceptor(traceCfg *primitive.TraceConfig) primitive.Interceptor
 
 		producerCtx := primitive.GetProducerCtx(ctx)
 		if producerCtx.Message.Topic == dispatcher.GetTraceTopicName() {
-			return next(ctx, req, reply)
+			return err
 		}
 
 		// SendOneway && SendAsync has no reply.
diff --git a/producer/producer.go b/producer/producer.go
index 8ebb660..06ea739 100644
--- a/producer/producer.go
+++ b/producer/producer.go
@@ -21,6 +21,7 @@ import (
 	"bytes"
 	"context"
 	"fmt"
+	errors2 "github.com/apache/rocketmq-client-go/v2/errors"
 	"strconv"
 	"sync"
 	"sync/atomic"
@@ -35,12 +36,6 @@ import (
 	"github.com/apache/rocketmq-client-go/v2/rlog"
 )
 
-var (
-	ErrTopicEmpty   = errors.New("topic is nil")
-	ErrMessageEmpty = errors.New("message is nil")
-	ErrNotRunning   = errors.New("producer not started")
-)
-
 type defaultProducer struct {
 	group       string
 	client      internal.RMQClient
@@ -72,7 +67,10 @@ func NewDefaultProducer(opts ...Option) (*defaultProducer, error) {
 		options:    defaultOpts,
 	}
 	producer.client = internal.GetOrNewRocketMQClient(defaultOpts.ClientOptions, producer.callbackCh)
-
+	producer.options.ClientOptions.Namesrv, err = internal.GetNamesrv(producer.client.ClientID())
+	if err != nil {
+		return nil, err
+	}
 	producer.interceptor = primitive.ChainInterceptors(producer.options.Interceptors...)
 
 	return producer, nil
@@ -95,15 +93,15 @@ func (p *defaultProducer) Shutdown() error {
 
 func (p *defaultProducer) checkMsg(msgs ...*primitive.Message) error {
 	if atomic.LoadInt32(&p.state) != int32(internal.StateRunning) {
-		return ErrNotRunning
+		return errors2.ErrNotRunning
 	}
 
 	if len(msgs) == 0 {
-		return errors.New("message is nil")
+		return errors2.ErrMessageEmpty
 	}
 
 	if len(msgs[0].Topic) == 0 {
-		return errors.New("topic is nil")
+		return errors2.ErrTopicEmpty
 	}
 	return nil
 }


Test patch:
diff --git a/internal/remote/remote_client_test.go b/internal/remote/remote_client_test.go
index f160674..fa33f4f 100644
--- a/internal/remote/remote_client_test.go
+++ b/internal/remote/remote_client_test.go
@@ -19,7 +19,7 @@ package remote
 import (
 	"bytes"
 	"context"
-	"errors"
+	"github.com/apache/rocketmq-client-go/v2/errors"
 	"math/rand"
 	"net"
 	"reflect"
@@ -27,8 +27,6 @@ import (
 	"testing"
 	"time"
 
-	"github.com/apache/rocketmq-client-go/v2/internal/utils"
-
 	"github.com/stretchr/testify/assert"
 )
 
@@ -80,12 +78,12 @@ func TestResponseFutureWaitResponse(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), time.Duration(1000))
 	defer cancel()
 	future := NewResponseFuture(ctx, 10, nil)
-	if _, err := future.waitResponse(); err != utils.ErrRequestTimeout {
+	if _, err := future.waitResponse(); err != errors.ErrRequestTimeout {
 		t.Errorf("wrong ResponseFuture waitResponse. want=%v, got=%v",
-			utils.ErrRequestTimeout, err)
+			errors.ErrRequestTimeout, err)
 	}
 	future = NewResponseFuture(context.Background(), 10, nil)
-	responseError := errors.New("response error")
+	responseError := errors.ErrResponse
 	go func() {
 		time.Sleep(100 * time.Millisecond)
 		future.Err = responseError
@@ -295,7 +293,7 @@ func TestInvokeAsyncTimeout(t *testing.T) {
 		err := client.InvokeAsync(ctx, addr, clientSendRemtingCommand,
 			func(r *ResponseFuture) {
 				assert.NotNil(t, r.Err)
-				assert.Equal(t, utils.ErrRequestTimeout, r.Err)
+				assert.Equal(t, errors.ErrRequestTimeout, r.Err)
 				wg.Done()
 			})
 		assert.Nil(t, err, "failed to invokeSync.")
diff --git a/internal/route_test.go b/internal/route_test.go
index ded7780..a1ebec4 100644
--- a/internal/route_test.go
+++ b/internal/route_test.go
@@ -19,11 +19,11 @@ package internal
 
 import (
 	"context"
+	"github.com/apache/rocketmq-client-go/v2/errors"
 	"sync"
 	"testing"
 
 	"github.com/golang/mock/gomock"
-	"github.com/pkg/errors"
 	. "github.com/smartystreets/goconvey/convey"
 	"github.com/stretchr/testify/assert"
 
@@ -53,7 +53,7 @@ func TestQueryTopicRouteInfoFromServer(t *testing.T) {
 				func(ctx context.Context, addr string, request *remote.RemotingCommand) (*remote.RemotingCommand, error) {
 					count++
 					if count < 3 {
-						return nil, errors.New("not existed")
+						return nil, errors.ErrNotExisted
 					}
 					return &remote.RemotingCommand{
 						Code: ResTopicNotExist,
@@ -62,7 +62,7 @@ func TestQueryTopicRouteInfoFromServer(t *testing.T) {
 
 			data, err := namesrv.queryTopicRouteInfoFromServer("notexisted")
 			assert.Nil(t, data)
-			assert.Equal(t, ErrTopicNotExist, err)
+			assert.Equal(t, errors.ErrTopicNotExist, err)
 		})
 	})
 }
diff --git a/producer/producer_test.go b/producer/producer_test.go
index 12cefb8..b6ec84d 100644
--- a/producer/producer_test.go
+++ b/producer/producer_test.go
@@ -19,6 +19,7 @@ package producer
 
 import (
 	"context"
+	"github.com/apache/rocketmq-client-go/v2/errors"
 	"testing"
 
 	"github.com/golang/mock/gomock"
@@ -60,17 +61,17 @@ func TestShutdown(t *testing.T) {
 	msg := new(primitive.Message)
 
 	r, err := p.SendSync(ctx, msg)
-	assert.Equal(t, ErrNotRunning, err)
+	assert.Equal(t, errors.ErrNotRunning, err)
 	assert.Nil(t, r)
 
 	err = p.SendOneWay(ctx, msg)
-	assert.Equal(t, ErrNotRunning, err)
+	assert.Equal(t, errors.ErrNotRunning, err)
 
 	f := func(context.Context, *primitive.SendResult, error) {
 		assert.False(t, true, "should not  come in")
 	}
 	err = p.SendAsync(ctx, f, msg)
-	assert.Equal(t, ErrNotRunning, err)
+	assert.Equal(t, errors.ErrNotRunning, err)
 }
 
 func mockB4Send(p *defaultProducer) {