Repository: apache/rocketmq-client-go
Pull request #732 (2021-10-14 13:22:07)

[ISSUE #731] Optimize the update offset logic
## What is the purpose of the change

#731 

## Brief changelog
![image](https://user-images.githubusercontent.com/30995057/137325708-4a4a38d6-0ef6-454f-97b5-2eaa07997daa.png)
![image](https://user-images.githubusercontent.com/30995057/137325723-62a1401e-abe0-47eb-a98f-ef53df7e2365.png)


## Verifying this change

XXXX

Follow this checklist to help us incorporate your contribution quickly and easily. Notice, `it would be helpful if you could finish the following 5 checklist(the last one is not necessary)before request the community to review your PR`.

- [x] Make sure there is a [Github issue](https://github.com/apache/rocketmq/issues) filed for the change (usually before you start working on it). Trivial changes like typos do not require a Github issue. Your pull request should address just this issue, without pulling in other changes - one PR resolves one issue. 
- [x] Format the pull request title like `[ISSUE #123] Fix UnknownException when host config not exist`. Each commit in the pull request should have a meaningful subject line and body.
- [x] Write a pull request description that is detailed enough to understand what the pull request does, how, and why.
- [x] Write necessary unit-test(over 80% coverage) to verify your logic correction, more mock a little better when a cross-module dependency exists.
- [ ] If this contribution is large, please file an [Apache Individual Contributor License Agreement](http://www.apache.org/licenses/#clas).



Patch:
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 9dfc62f..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 {
diff --git a/consumer/pull_consumer.go b/consumer/pull_consumer.go
index 63e9c54..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"
 )
@@ -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 66bf367..9b38d0f 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"
@@ -147,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
 		}
 
@@ -233,7 +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.")
+		return errors2.ErrStartTopic
 	}
 
 	if pc.option.Namespace != "" {
@@ -763,10 +764,7 @@ func (pc *pushConsumer) pullMessage(request *PullRequest) {
 					"prevRequestOffset": prevRequestOffset,
 				})
 			}
-		case primitive.PullNoNewMsg:
-			rlog.Debug(fmt.Sprintf("Topic: %s, QueueId: %d no more msg, current offset: %d, next offset: %d",
-				request.mq.Topic, request.mq.QueueId, pullRequest.QueueOffset, result.NextBeginOffset), nil)
-		case primitive.PullNoMsgMatched:
+		case primitive.PullNoNewMsg, primitive.PullNoMsgMatched:
 			request.nextOffset = result.NextBeginOffset
 			pc.correctTagsOffset(request)
 		case primitive.PullOffsetIllegal:
@@ -789,7 +787,9 @@ func (pc *pushConsumer) pullMessage(request *PullRequest) {
 }
 
 func (pc *pushConsumer) correctTagsOffset(pr *PullRequest) {
-	// TODO
+	if pr.pq.cachedMsgCount <= 0 {
+		pc.storage.update(pr.mq, pr.nextOffset, true)
+	}
 }
 
 func (pc *pushConsumer) sendMessageBack(brokerName string, msg *primitive.MessageExt, delayLevel int) bool {
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/internal/client.go b/internal/client.go
index 4de3b93..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
 )
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 1fbc48b..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()
@@ -434,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/producer.go b/producer/producer.go
index ef4ea25..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
@@ -98,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) {