Repository: apache/rocketmq-client-go
Pull request #773 (2022-01-23 12:40:48)

fix(sendOneWay): fix stack overflow caused by recursive calls
## What is the purpose of the change

fix stack overflow caused by recursive calls

![image](https://user-images.githubusercontent.com/19374680/150678843-e4974073-8fca-4bdd-8caf-e6c9380360d2.png)

![image](https://user-images.githubusercontent.com/19374680/150678790-b104b874-6712-41e6-9d37-ab08bb100ba0.png)

## Brief changelog


## Verifying this change

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/admin/admin.go b/admin/admin.go
index 1957a06..8169dcb 100644
--- a/admin/admin.go
+++ b/admin/admin.go
@@ -19,6 +19,7 @@ package admin
 
 import (
 	"context"
+	"fmt"
 	"sync"
 	"time"
 
@@ -61,8 +62,7 @@ func WithResolver(resolver primitive.NsResolver) AdminOption {
 }
 
 type admin struct {
-	cli     internal.RMQClient
-	namesrv internal.Namesrvs
+	cli internal.RMQClient
 
 	opts *adminOptions
 
@@ -75,17 +75,21 @@ func NewAdmin(opts ...AdminOption) (Admin, error) {
 	for _, opt := range opts {
 		opt(defaultOpts)
 	}
-
-	cli := internal.GetOrNewRocketMQClient(defaultOpts.ClientOptions, nil)
-	namesrv, err := internal.NewNamesrv(defaultOpts.Resolver)
+	namesrv, err := internal.NewNamesrv(defaultOpts.Resolver, defaultOpts.RemotingClientConfig)
+	defaultOpts.Namesrv = namesrv
 	if err != nil {
 		return nil, err
 	}
+
+	cli := internal.GetOrNewRocketMQClient(defaultOpts.ClientOptions, nil)
+	if cli == nil {
+		return nil, fmt.Errorf("GetOrNewRocketMQClient faild")
+	}
+	defaultOpts.Namesrv = cli.GetNameSrv()
 	//log.Printf("Client: %#v", namesrv.srvs)
 	return &admin{
-		cli:     cli,
-		namesrv: namesrv,
-		opts:    defaultOpts,
+		cli:  cli,
+		opts: defaultOpts,
 	}, nil
 }
 
@@ -153,8 +157,8 @@ func (a *admin) DeleteTopic(ctx context.Context, opts ...OptionDelete) error {
 	}
 	//delete topic in broker
 	if cfg.BrokerAddr == "" {
-		a.namesrv.UpdateTopicRouteInfo(cfg.Topic)
-		cfg.BrokerAddr = a.namesrv.FindBrokerAddrByTopic(cfg.Topic)
+		a.cli.GetNameSrv().UpdateTopicRouteInfo(cfg.Topic)
+		cfg.BrokerAddr = a.cli.GetNameSrv().FindBrokerAddrByTopic(cfg.Topic)
 	}
 
 	if _, err := a.deleteTopicInBroker(ctx, cfg.Topic, cfg.BrokerAddr); err != nil {
@@ -168,14 +172,16 @@ func (a *admin) DeleteTopic(ctx context.Context, opts ...OptionDelete) error {
 
 	//delete topic in nameserver
 	if len(cfg.NameSrvAddr) == 0 {
-		_, _, err := a.namesrv.UpdateTopicRouteInfo(cfg.Topic)
+		a.cli.GetNameSrv().UpdateTopicRouteInfo(cfg.Topic)
+		cfg.NameSrvAddr = a.cli.GetNameSrv().AddrList()
+		_, _, err := a.cli.GetNameSrv().UpdateTopicRouteInfo(cfg.Topic)
 		if err != nil {
 			rlog.Error("delete topic in nameserver error", map[string]interface{}{
-				rlog.LogKeyTopic: cfg.Topic,
+				rlog.LogKeyTopic:         cfg.Topic,
 				rlog.LogKeyUnderlayError: err,
 			})
 		}
-		cfg.NameSrvAddr = a.namesrv.AddrList()
+		cfg.NameSrvAddr = a.cli.GetNameSrv().AddrList()
 	}
 
 	for _, nameSrvAddr := range cfg.NameSrvAddr {
diff --git a/api.go b/api.go
index 31f58d5..8d6d781 100644
--- a/api.go
+++ b/api.go
@@ -19,9 +19,11 @@ package rocketmq
 
 import (
 	"context"
-	"github.com/apache/rocketmq-client-go/v2/errors"
+	"time"
 
 	"github.com/apache/rocketmq-client-go/v2/consumer"
+	"github.com/apache/rocketmq-client-go/v2/errors"
+	"github.com/apache/rocketmq-client-go/v2/internal"
 	"github.com/apache/rocketmq-client-go/v2/primitive"
 	"github.com/apache/rocketmq-client-go/v2/producer"
 )
@@ -33,6 +35,8 @@ type Producer interface {
 	SendAsync(ctx context.Context, mq func(ctx context.Context, result *primitive.SendResult, err error),
 		msg ...*primitive.Message) error
 	SendOneWay(ctx context.Context, mq ...*primitive.Message) error
+	Request(ctx context.Context, ttl time.Duration, msg *primitive.Message) (*primitive.Message, error)
+	RequestAsync(ctx context.Context, ttl time.Duration, callback internal.RequestCallback, msg *primitive.Message) error
 }
 
 func NewProducer(opts ...producer.Option) (Producer, error) {
diff --git a/benchmark/consumer.go b/benchmark/consumer.go
index 907a1e7..a331760 100644
--- a/benchmark/consumer.go
+++ b/benchmark/consumer.go
@@ -91,11 +91,11 @@ func (s *consumeSnapshots) printStati() {
 	s.RUnlock()
 
 	rlog.Info("Benchmark Consumer Snapshot", map[string]interface{}{
-		"consumeTPS": int64(consumeTps),
+		"consumeTPS":     int64(consumeTps),
 		"average(B2C)RT": avgB2CRT,
 		"average(S2C)RT": avgS2CRT,
-		"max(B2C)RT": l.born2ConsumerMaxRT,
-		"max(S2C)RT": l.store2ConsumerMaxRT,
+		"max(B2C)RT":     l.born2ConsumerMaxRT,
+		"max(S2C)RT":     l.store2ConsumerMaxRT,
 	})
 }
 
diff --git a/benchmark/producer.go b/benchmark/producer.go
index 7516352..8125bcf 100644
--- a/benchmark/producer.go
+++ b/benchmark/producer.go
@@ -92,12 +92,12 @@ func (s *produceSnapshots) printStati() {
 	s.RUnlock()
 
 	rlog.Info("Benchmark Producer Snapshot", map[string]interface{}{
-		"sendTps": int64(sendTps),
-		"maxRt": maxRT,
-		"averageRt": avgRT,
-		"sendFailed": l.sendRequestFailedCount,
+		"sendTps":        int64(sendTps),
+		"maxRt":          maxRT,
+		"averageRt":      avgRT,
+		"sendFailed":     l.sendRequestFailedCount,
 		"responseFailed": l.receiveResponseFailedCount,
-		"total": l.receiveResponseSuccessCount,
+		"total":          l.receiveResponseSuccessCount,
 	})
 }
 
@@ -179,8 +179,8 @@ AGAIN:
 		goto AGAIN
 	}
 	rlog.Error("Send Message Error", map[string]interface{}{
-		"topic": topic,
-		"tag": tag,
+		"topic":                  topic,
+		"tag":                    tag,
 		rlog.LogKeyUnderlayError: err.Error(),
 	})
 	goto AGAIN
diff --git a/benchmark/stable.go b/benchmark/stable.go
index 2659bc5..43525b3 100644
--- a/benchmark/stable.go
+++ b/benchmark/stable.go
@@ -128,7 +128,7 @@ func (stp *stableTestProducer) run(args []string) {
 	err := stp.flags.Parse(args)
 	if err != nil {
 		rlog.Info("Parse Args Error", map[string]interface{}{
-			"args": args,
+			"args":                   args,
 			rlog.LogKeyUnderlayError: err.Error(),
 		})
 		stp.usage()
@@ -205,7 +205,7 @@ func (stc *stableTestConsumer) run(args []string) {
 	err := stc.flags.Parse(args)
 	if err != nil {
 		rlog.Error("Parse Args Error", map[string]interface{}{
-			"args": args,
+			"args":                   args,
 			rlog.LogKeyUnderlayError: err.Error(),
 		})
 		stc.usage()
diff --git a/consumer/consumer.go b/consumer/consumer.go
index ad77bc1..6e53eaf 100644
--- a/consumer/consumer.go
+++ b/consumer/consumer.go
@@ -20,7 +20,6 @@ package consumer
 import (
 	"context"
 	"fmt"
-	"github.com/apache/rocketmq-client-go/v2/errors"
 	"sort"
 	"strconv"
 	"strings"
@@ -28,8 +27,9 @@ import (
 	"sync/atomic"
 	"time"
 
-	jsoniter "github.com/json-iterator/go"
+	"github.com/apache/rocketmq-client-go/v2/errors"
 
+	jsoniter "github.com/json-iterator/go"
 	"github.com/tidwall/gjson"
 
 	"github.com/apache/rocketmq-client-go/v2/internal"
@@ -263,8 +263,6 @@ type defaultConsumer struct {
 	// chan for push consumer
 	prCh chan PullRequest
 
-	namesrv internal.Namesrvs
-
 	pullFromWhichNodeTable sync.Map
 
 	stat *StatsManager
@@ -280,7 +278,7 @@ func (dc *defaultConsumer) start() error {
 
 	if dc.model == Clustering {
 		dc.option.ChangeInstanceNameToPID()
-		dc.storage = NewRemoteOffsetStore(dc.consumerGroup, dc.client, dc.namesrv)
+		dc.storage = NewRemoteOffsetStore(dc.consumerGroup, dc.client, dc.client.GetNameSrv())
 	} else {
 		dc.storage = NewLocalFileOffsetStore(dc.consumerGroup, dc.client.ClientID())
 	}
@@ -448,7 +446,7 @@ type lockBatchRequestBody struct {
 }
 
 func (dc *defaultConsumer) lock(mq *primitive.MessageQueue) bool {
-	brokerResult := dc.namesrv.FindBrokerAddressInSubscribe(mq.BrokerName, internal.MasterId, true)
+	brokerResult := dc.client.GetNameSrv().FindBrokerAddressInSubscribe(mq.BrokerName, internal.MasterId, true)
 
 	if brokerResult == nil {
 		return false
@@ -488,7 +486,7 @@ func (dc *defaultConsumer) lock(mq *primitive.MessageQueue) bool {
 }
 
 func (dc *defaultConsumer) unlock(mq *primitive.MessageQueue, oneway bool) {
-	brokerResult := dc.namesrv.FindBrokerAddressInSubscribe(mq.BrokerName, internal.MasterId, true)
+	brokerResult := dc.client.GetNameSrv().FindBrokerAddressInSubscribe(mq.BrokerName, internal.MasterId, true)
 
 	if brokerResult == nil {
 		return
@@ -513,7 +511,7 @@ func (dc *defaultConsumer) lockAll() {
 		if len(mqs) == 0 {
 			continue
 		}
-		brokerResult := dc.namesrv.FindBrokerAddressInSubscribe(broker, internal.MasterId, true)
+		brokerResult := dc.client.GetNameSrv().FindBrokerAddressInSubscribe(broker, internal.MasterId, true)
 		if brokerResult == nil {
 			continue
 		}
@@ -559,7 +557,7 @@ func (dc *defaultConsumer) unlockAll(oneway bool) {
 		if len(mqs) == 0 {
 			continue
 		}
-		brokerResult := dc.namesrv.FindBrokerAddressInSubscribe(broker, internal.MasterId, true)
+		brokerResult := dc.client.GetNameSrv().FindBrokerAddressInSubscribe(broker, internal.MasterId, true)
 		if brokerResult == nil {
 			continue
 		}
@@ -702,8 +700,9 @@ func (dc *defaultConsumer) updateProcessQueueTable(topic string, mqs []*primitiv
 				continue
 			}
 			dc.storage.remove(&mq)
-			nextOffset := dc.computePullFromWhere(&mq)
-			if nextOffset >= 0 {
+			nextOffset, err := dc.computePullFromWhereWithException(&mq)
+
+			if nextOffset >= 0 && err == nil {
 				_, exist := dc.processQueueTable.Load(mq)
 				if exist {
 					rlog.Debug("do defaultConsumer, mq already exist", map[string]interface{}{
@@ -744,12 +743,23 @@ func (dc *defaultConsumer) removeUnnecessaryMessageQueue(mq *primitive.MessageQu
 	return true
 }
 
+// Deprecated: Use computePullFromWhereWithException instead.
 func (dc *defaultConsumer) computePullFromWhere(mq *primitive.MessageQueue) int64 {
+	result, _ := dc.computePullFromWhereWithException(mq)
+	return result
+}
+
+func (dc *defaultConsumer) computePullFromWhereWithException(mq *primitive.MessageQueue) (int64, error) {
 	if dc.cType == _PullConsume {
-		return 0
+		return 0, nil
+	}
+	result := int64(-1)
+	lastOffset, err := dc.storage.readWithException(mq, _ReadFromStore)
+	if err != nil {
+		// 这里 lastOffset = -1
+		return lastOffset, err
 	}
-	var result = int64(-1)
-	lastOffset := dc.storage.read(mq, _ReadFromStore)
+
 	if lastOffset >= 0 {
 		result = lastOffset
 	} else {
@@ -806,7 +816,7 @@ func (dc *defaultConsumer) computePullFromWhere(mq *primitive.MessageQueue) int6
 		default:
 		}
 	}
-	return result
+	return result, nil
 }
 
 func (dc *defaultConsumer) pullInner(ctx context.Context, queue *primitive.MessageQueue, data *internal.SubscriptionData,
@@ -892,10 +902,10 @@ func (dc *defaultConsumer) processPullResult(mq *primitive.MessageQueue, result
 }
 
 func (dc *defaultConsumer) findConsumerList(topic string) []string {
-	brokerAddr := dc.namesrv.FindBrokerAddrByTopic(topic)
+	brokerAddr := dc.client.GetNameSrv().FindBrokerAddrByTopic(topic)
 	if brokerAddr == "" {
-		dc.namesrv.UpdateTopicRouteInfo(topic)
-		brokerAddr = dc.namesrv.FindBrokerAddrByTopic(topic)
+		dc.client.GetNameSrv().UpdateTopicRouteInfo(topic)
+		brokerAddr = dc.client.GetNameSrv().FindBrokerAddrByTopic(topic)
 	}
 
 	if brokerAddr != "" {
@@ -929,10 +939,10 @@ func (dc *defaultConsumer) sendBack(msg *primitive.MessageExt, level int) error
 
 // QueryMaxOffset with specific queueId and topic
 func (dc *defaultConsumer) queryMaxOffset(mq *primitive.MessageQueue) (int64, error) {
-	brokerAddr := dc.namesrv.FindBrokerAddrByName(mq.BrokerName)
+	brokerAddr := dc.client.GetNameSrv().FindBrokerAddrByName(mq.BrokerName)
 	if brokerAddr == "" {
-		dc.namesrv.UpdateTopicRouteInfo(mq.Topic)
-		brokerAddr = dc.namesrv.FindBrokerAddrByName(mq.BrokerName)
+		dc.client.GetNameSrv().UpdateTopicRouteInfo(mq.Topic)
+		brokerAddr = dc.client.GetNameSrv().FindBrokerAddrByName(mq.BrokerName)
 	}
 	if brokerAddr == "" {
 		return -1, fmt.Errorf("the broker [%s] does not exist", mq.BrokerName)
@@ -953,15 +963,16 @@ func (dc *defaultConsumer) queryMaxOffset(mq *primitive.MessageQueue) (int64, er
 }
 
 func (dc *defaultConsumer) queryOffset(mq *primitive.MessageQueue) int64 {
-	return dc.storage.read(mq, _ReadMemoryThenStore)
+	result, _ := dc.storage.readWithException(mq, _ReadMemoryThenStore)
+	return result
 }
 
 // SearchOffsetByTimestamp with specific queueId and topic
 func (dc *defaultConsumer) searchOffsetByTimestamp(mq *primitive.MessageQueue, timestamp int64) (int64, error) {
-	brokerAddr := dc.namesrv.FindBrokerAddrByName(mq.BrokerName)
+	brokerAddr := dc.client.GetNameSrv().FindBrokerAddrByName(mq.BrokerName)
 	if brokerAddr == "" {
-		dc.namesrv.UpdateTopicRouteInfo(mq.Topic)
-		brokerAddr = dc.namesrv.FindBrokerAddrByName(mq.BrokerName)
+		dc.client.GetNameSrv().UpdateTopicRouteInfo(mq.Topic)
+		brokerAddr = dc.client.GetNameSrv().FindBrokerAddrByName(mq.BrokerName)
 	}
 	if brokerAddr == "" {
 		return -1, fmt.Errorf("the broker [%s] does not exist", mq.BrokerName)
@@ -1044,12 +1055,12 @@ func clearCommitOffsetFlag(sysFlag int32) int32 {
 }
 
 func (dc *defaultConsumer) tryFindBroker(mq *primitive.MessageQueue) *internal.FindBrokerResult {
-	result := dc.namesrv.FindBrokerAddressInSubscribe(mq.BrokerName, dc.recalculatePullFromWhichNode(mq), false)
+	result := dc.client.GetNameSrv().FindBrokerAddressInSubscribe(mq.BrokerName, dc.recalculatePullFromWhichNode(mq), false)
 	if result != nil {
 		return result
 	}
-	dc.namesrv.UpdateTopicRouteInfo(mq.Topic)
-	return dc.namesrv.FindBrokerAddressInSubscribe(mq.BrokerName, dc.recalculatePullFromWhichNode(mq), false)
+	dc.client.GetNameSrv().UpdateTopicRouteInfo(mq.Topic)
+	return dc.client.GetNameSrv().FindBrokerAddressInSubscribe(mq.BrokerName, dc.recalculatePullFromWhichNode(mq), false)
 }
 
 func (dc *defaultConsumer) updatePullFromWhichNode(mq *primitive.MessageQueue, brokerId int64) {
diff --git a/consumer/interceptor.go b/consumer/interceptor.go
index aababfe..05ff94a 100644
--- a/consumer/interceptor.go
+++ b/consumer/interceptor.go
@@ -19,6 +19,7 @@ package consumer
 
 import (
 	"context"
+	"fmt"
 	"time"
 
 	"github.com/apache/rocketmq-client-go/v2/internal"
@@ -39,9 +40,14 @@ func WithTrace(traceCfg *primitive.TraceConfig) Option {
 
 func newTraceInterceptor(traceCfg *primitive.TraceConfig) primitive.Interceptor {
 	dispatcher := internal.NewTraceDispatcher(traceCfg)
-	dispatcher.Start()
+	if dispatcher != nil {
+		dispatcher.Start()
+	}
 
 	return func(ctx context.Context, req, reply interface{}, next primitive.Invoker) error {
+		if dispatcher == nil {
+			return fmt.Errorf("GetOrNewRocketMQClient faild")
+		}
 		consumerCtx, exist := primitive.GetConsumerCtx(ctx)
 		if !exist || len(consumerCtx.Msgs) == 0 {
 			return next(ctx, req, reply)
diff --git a/consumer/mock_offset_store.go b/consumer/mock_offset_store.go
index bac1cdb..145ec3d 100644
--- a/consumer/mock_offset_store.go
+++ b/consumer/mock_offset_store.go
@@ -28,67 +28,90 @@ import (
 	gomock "github.com/golang/mock/gomock"
 )
 
-// MockOffsetStore is a mock of OffsetStore interface
+// MockOffsetStore is a mock of OffsetStore interface.
 type MockOffsetStore struct {
 	ctrl     *gomock.Controller
 	recorder *MockOffsetStoreMockRecorder
 }
 
-// MockOffsetStoreMockRecorder is the mock recorder for MockOffsetStore
+// MockOffsetStoreMockRecorder is the mock recorder for MockOffsetStore.
 type MockOffsetStoreMockRecorder struct {
 	mock *MockOffsetStore
 }
 
-// NewMockOffsetStore creates a new mock instance
+// NewMockOffsetStore creates a new mock instance.
 func NewMockOffsetStore(ctrl *gomock.Controller) *MockOffsetStore {
 	mock := &MockOffsetStore{ctrl: ctrl}
 	mock.recorder = &MockOffsetStoreMockRecorder{mock}
 	return mock
 }
 
-// EXPECT returns an object that allows the caller to indicate expected use
+// EXPECT returns an object that allows the caller to indicate expected use.
 func (m *MockOffsetStore) EXPECT() *MockOffsetStoreMockRecorder {
 	return m.recorder
 }
 
-// persist mocks base method
+// persist mocks base method.
 func (m *MockOffsetStore) persist(mqs []*primitive.MessageQueue) {
+	m.ctrl.T.Helper()
 	m.ctrl.Call(m, "persist", mqs)
 }
 
-// persist indicates an expected call of persist
+// persist indicates an expected call of persist.
 func (mr *MockOffsetStoreMockRecorder) persist(mqs interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "persist", reflect.TypeOf((*MockOffsetStore)(nil).persist), mqs)
 }
 
-// remove mocks base method
-func (m *MockOffsetStore) remove(mq *primitive.MessageQueue) {
-	m.ctrl.Call(m, "remove", mq)
-}
-
-// remove indicates an expected call of remove
-func (mr *MockOffsetStoreMockRecorder) remove(mq interface{}) *gomock.Call {
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "remove", reflect.TypeOf((*MockOffsetStore)(nil).remove), mq)
-}
-
-// read mocks base method
+// read mocks base method.
 func (m *MockOffsetStore) read(mq *primitive.MessageQueue, t readType) int64 {
+	m.ctrl.T.Helper()
 	ret := m.ctrl.Call(m, "read", mq, t)
 	ret0, _ := ret[0].(int64)
 	return ret0
 }
 
-// read indicates an expected call of read
+// read indicates an expected call of read.
 func (mr *MockOffsetStoreMockRecorder) read(mq, t interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "read", reflect.TypeOf((*MockOffsetStore)(nil).read), mq, t)
 }
 
-// update mocks base method
+// readWithException mocks base method.
+func (m *MockOffsetStore) readWithException(mq *primitive.MessageQueue, t readType) (int64, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "readWithException", mq, t)
+	ret0, _ := ret[0].(int64)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// readWithException indicates an expected call of readWithException.
+func (mr *MockOffsetStoreMockRecorder) readWithException(mq, t interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "readWithException", reflect.TypeOf((*MockOffsetStore)(nil).readWithException), mq, t)
+}
+
+// remove mocks base method.
+func (m *MockOffsetStore) remove(mq *primitive.MessageQueue) {
+	m.ctrl.T.Helper()
+	m.ctrl.Call(m, "remove", mq)
+}
+
+// remove indicates an expected call of remove.
+func (mr *MockOffsetStoreMockRecorder) remove(mq interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "remove", reflect.TypeOf((*MockOffsetStore)(nil).remove), mq)
+}
+
+// update mocks base method.
 func (m *MockOffsetStore) update(mq *primitive.MessageQueue, offset int64, increaseOnly bool) {
+	m.ctrl.T.Helper()
 	m.ctrl.Call(m, "update", mq, offset, increaseOnly)
 }
 
-// update indicates an expected call of update
+// update indicates an expected call of update.
 func (mr *MockOffsetStoreMockRecorder) update(mq, offset, increaseOnly interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "update", reflect.TypeOf((*MockOffsetStore)(nil).update), mq, offset, increaseOnly)
 }
diff --git a/consumer/offset_store.go b/consumer/offset_store.go
index 5ecfd14..86ecd18 100644
--- a/consumer/offset_store.go
+++ b/consumer/offset_store.go
@@ -57,6 +57,7 @@ type OffsetStore interface {
 	persist(mqs []*primitive.MessageQueue)
 	remove(mq *primitive.MessageQueue)
 	read(mq *primitive.MessageQueue, t readType) int64
+	readWithException(mq *primitive.MessageQueue, t readType) (int64, error)
 	update(mq *primitive.MessageQueue, offset int64, increaseOnly bool)
 }
 
@@ -156,21 +157,27 @@ func (local *localFileOffsetStore) load() {
 	}
 }
 
+// Deprecated: Use readWithException instead.
 func (local *localFileOffsetStore) read(mq *primitive.MessageQueue, t readType) int64 {
+	result, _ := local.readWithException(mq, t)
+	return result
+}
+
+func (local *localFileOffsetStore) readWithException(mq *primitive.MessageQueue, t readType) (int64, error) {
 	switch t {
 	case _ReadFromMemory, _ReadMemoryThenStore:
 		off := readFromMemory(local.OffsetTable, mq)
 		if off >= 0 || (off == -1 && t == _ReadFromMemory) {
-			return off
+			return off, nil
 		}
 		fallthrough
 	case _ReadFromStore:
 		local.load()
-		return readFromMemory(local.OffsetTable, mq)
+		return readFromMemory(local.OffsetTable, mq), nil
 	default:
 
 	}
-	return -1
+	return -1, nil
 }
 
 func (local *localFileOffsetStore) update(mq *primitive.MessageQueue, offset int64, increaseOnly bool) {
@@ -284,18 +291,24 @@ func (r *remoteBrokerOffsetStore) remove(mq *primitive.MessageQueue) {
 	})
 }
 
+// Deprecated: Use readWithException instead.
 func (r *remoteBrokerOffsetStore) read(mq *primitive.MessageQueue, t readType) int64 {
+	result, _ := r.readWithException(mq, t)
+	return result
+}
+
+func (r *remoteBrokerOffsetStore) readWithException(mq *primitive.MessageQueue, t readType) (int64, error) {
 	r.mutex.RLock()
 	switch t {
 	case _ReadFromMemory, _ReadMemoryThenStore:
 		off, exist := r.OffsetTable[*mq]
 		if exist {
 			r.mutex.RUnlock()
-			return off
+			return off, nil
 		}
 		if t == _ReadFromMemory {
 			r.mutex.RUnlock()
-			return -1
+			return -1, nil
 		}
 		fallthrough
 	case _ReadFromStore:
@@ -307,7 +320,7 @@ func (r *remoteBrokerOffsetStore) read(mq *primitive.MessageQueue, t readType) i
 				rlog.LogKeyUnderlayError: err,
 			})
 			r.mutex.RUnlock()
-			return -1
+			return -1, err
 		}
 		rlog.Warning("fetch offset of mq from broker success", map[string]interface{}{
 			rlog.LogKeyConsumerGroup: r.group,
@@ -316,11 +329,11 @@ func (r *remoteBrokerOffsetStore) read(mq *primitive.MessageQueue, t readType) i
 		})
 		r.mutex.RUnlock()
 		r.update(mq, off, true)
-		return off
+		return off, nil
 	default:
 	}
 
-	return -1
+	return -1, nil
 }
 
 func (r *remoteBrokerOffsetStore) update(mq *primitive.MessageQueue, offset int64, increaseOnly bool) {
diff --git a/consumer/option.go b/consumer/option.go
index 330f6c4..1ef184b 100644
--- a/consumer/option.go
+++ b/consumer/option.go
@@ -118,6 +118,7 @@ func defaultPushConsumerOptions() consumerOptions {
 		ConsumerModel:              Clustering,
 		AutoCommit:                 true,
 		Resolver:                   primitive.NewHttpResolver("DEFAULT"),
+		ConsumeTimestamp: 			time.Now().Add(time.Minute * (-30)).Format("20060102150405"),
 	}
 	opts.ClientOptions.GroupName = "DEFAULT_CONSUMER"
 	return opts
@@ -158,6 +159,24 @@ func WithConsumeMessageBatchMaxSize(consumeMessageBatchMaxSize int) Option {
 	}
 }
 
+func WithConsumeTimestamp(consumeTimestamp string) Option {
+	return func(options *consumerOptions) {
+		options.ConsumeTimestamp = consumeTimestamp
+	}
+}
+
+func WithConsumerPullTimeout(consumerPullTimeout time.Duration) Option {
+	return func(options *consumerOptions) {
+		options.ConsumerPullTimeout = consumerPullTimeout
+	}
+}
+
+func WithConsumeConcurrentlyMaxSpan(consumeConcurrentlyMaxSpan int) Option {
+	return func(options *consumerOptions) {
+		options.ConsumeConcurrentlyMaxSpan = consumeConcurrentlyMaxSpan
+	}
+}
+
 // WithChainConsumerInterceptor returns a ConsumerOption that specifies the chained interceptor for consumer.
 // The first interceptor will be the outer most, while the last interceptor will be the inner most wrapper
 // around the real call.
diff --git a/consumer/pull_consumer.go b/consumer/pull_consumer.go
index 81a3ec5..4699601 100644
--- a/consumer/pull_consumer.go
+++ b/consumer/pull_consumer.go
@@ -20,10 +20,11 @@ package consumer
 import (
 	"context"
 	"fmt"
-	errors2 "github.com/apache/rocketmq-client-go/v2/errors"
 	"sync"
 	"sync/atomic"
 
+	errors2 "github.com/apache/rocketmq-client-go/v2/errors"
+
 	"github.com/pkg/errors"
 
 	"github.com/apache/rocketmq-client-go/v2/internal"
@@ -76,7 +77,7 @@ func NewPullConsumer(options ...Option) (*defaultPullConsumer, error) {
 		apply(&defaultOpts)
 	}
 
-	srvs, err := internal.NewNamesrv(defaultOpts.Resolver)
+	srvs, err := internal.NewNamesrv(defaultOpts.Resolver, defaultOpts.RemotingClientConfig)
 	if err != nil {
 		return nil, errors.Wrap(err, "new Namesrv failed.")
 	}
@@ -90,10 +91,12 @@ func NewPullConsumer(options ...Option) (*defaultPullConsumer, error) {
 		prCh:          make(chan PullRequest, 4),
 		model:         defaultOpts.ConsumerModel,
 		option:        defaultOpts,
-
-		namesrv: srvs,
 	}
-	dc.option.ClientOptions.Namesrv, err = internal.GetNamesrv(dc.client.ClientID())
+	if dc.client == nil {
+		return nil, fmt.Errorf("GetOrNewRocketMQClient faild")
+	}
+	defaultOpts.Namesrv = dc.client.GetNameSrv()
+
 	c := &defaultPullConsumer{
 		defaultConsumer: dc,
 	}
@@ -132,7 +135,7 @@ func (c *defaultPullConsumer) Pull(ctx context.Context, topic string, selector M
 }
 
 func (c *defaultPullConsumer) getNextQueueOf(topic string) *primitive.MessageQueue {
-	queues, err := c.defaultConsumer.namesrv.FetchSubscribeMessageQueues(topic)
+	queues, err := c.defaultConsumer.client.GetNameSrv().FetchSubscribeMessageQueues(topic)
 	if err != nil && len(queues) > 0 {
 		rlog.Error("get next mq error", map[string]interface{}{
 			rlog.LogKeyTopic:         topic,
@@ -217,7 +220,8 @@ func (c *defaultPullConsumer) makeSureStateOK() error {
 }
 
 func (c *defaultPullConsumer) nextOffsetOf(queue *primitive.MessageQueue) int64 {
-	return c.computePullFromWhere(queue)
+	result, _ := c.computePullFromWhereWithException(queue)
+	return result
 }
 
 // PullFrom pull messages of queue from the offset to offset + numbers
diff --git a/consumer/push_consumer.go b/consumer/push_consumer.go
index 3e4377b..4ad5ee3 100644
--- a/consumer/push_consumer.go
+++ b/consumer/push_consumer.go
@@ -20,14 +20,16 @@ package consumer
 import (
 	"context"
 	"fmt"
-	errors2 "github.com/apache/rocketmq-client-go/v2/errors"
 	"math"
+	"runtime/pprof"
 	"strconv"
 	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
 
+	errors2 "github.com/apache/rocketmq-client-go/v2/errors"
+
 	"github.com/pkg/errors"
 
 	"github.com/apache/rocketmq-client-go/v2/internal"
@@ -77,7 +79,7 @@ func NewPushConsumer(opts ...Option) (*pushConsumer, error) {
 	for _, apply := range opts {
 		apply(&defaultOpts)
 	}
-	srvs, err := internal.NewNamesrv(defaultOpts.Resolver)
+	srvs, err := internal.NewNamesrv(defaultOpts.Resolver, defaultOpts.RemotingClientConfig)
 	if err != nil {
 		return nil, errors.Wrap(err, "new Namesrv failed.")
 	}
@@ -100,14 +102,13 @@ func NewPushConsumer(opts ...Option) (*pushConsumer, error) {
 		consumeOrderly: defaultOpts.ConsumeOrderly,
 		fromWhere:      defaultOpts.FromWhere,
 		allocate:       defaultOpts.Strategy,
-		namesrv:        srvs,
 		option:         defaultOpts,
 	}
-	dc.option.ClientOptions.Namesrv, err = internal.GetNamesrv(dc.client.ClientID())
-	if err != nil {
-		return nil, err
+	if dc.client == nil {
+		return nil, fmt.Errorf("GetOrNewRocketMQClient faild")
 	}
-	dc.namesrv = dc.option.ClientOptions.Namesrv
+	defaultOpts.Namesrv = dc.client.GetNameSrv()
+
 	p := &pushConsumer{
 		defaultConsumer: dc,
 		subscribedTopic: make(map[string]string, 0),
@@ -124,11 +125,6 @@ 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
 }
 
@@ -358,7 +354,7 @@ func (pc *pushConsumer) ConsumeMessageDirectly(msg *primitive.MessageExt, broker
 	return res
 }
 
-func (pc *pushConsumer) GetConsumerRunningInfo() *internal.ConsumerRunningInfo {
+func (pc *pushConsumer) GetConsumerRunningInfo(stack bool) *internal.ConsumerRunningInfo {
 	info := internal.NewConsumerRunningInfo()
 
 	pc.subscriptionDataTable.Range(func(key, value interface{}) bool {
@@ -380,13 +376,26 @@ func (pc *pushConsumer) GetConsumerRunningInfo() *internal.ConsumerRunningInfo {
 		mq := key.(primitive.MessageQueue)
 		pq := value.(*processQueue)
 		pInfo := pq.currentInfo()
-		pInfo.CommitOffset = pc.storage.read(&mq, _ReadMemoryThenStore)
+		pInfo.CommitOffset, _ = pc.storage.readWithException(&mq, _ReadMemoryThenStore)
 		info.MQTable[mq] = pInfo
 		return true
 	})
 
+	if stack {
+		var buffer strings.Builder
+
+		err := pprof.Lookup("goroutine").WriteTo(&buffer, 2)
+		if err != nil {
+			rlog.Error("error when get stack ", map[string]interface{}{
+				"error": err,
+			})
+		} else {
+			info.JStack = buffer.String()
+		}
+	}
+
 	nsAddr := ""
-	for _, value := range pc.namesrv.AddrList() {
+	for _, value := range pc.client.GetNameSrv().AddrList() {
 		nsAddr += fmt.Sprintf("%s;", value)
 	}
 	info.Properties[internal.PropNameServerAddr] = nsAddr
@@ -629,13 +638,22 @@ func (pc *pushConsumer) pullMessage(request *PullRequest) {
 						rlog.LogKeyPullRequest:       request.String(),
 					})
 				}
+				pc.queueMaxSpanFlowControlTimes++
 				sleepTime = _PullDelayTimeWhenFlowControl
 				goto NEXT
 			}
 		} else {
 			if pq.IsLock() {
 				if !request.lockedFirst {
-					offset := pc.computePullFromWhere(request.mq)
+					offset, err := pc.computePullFromWhereWithException(request.mq)
+					if err != nil {
+						rlog.Warning("computePullFromWhere from broker error", map[string]interface{}{
+							rlog.LogKeyUnderlayError: err.Error(),
+						})
+						sleepTime = _PullDelayTimeWhenError
+						goto NEXT
+					}
+
 					brokerBusy := offset < request.nextOffset
 					rlog.Info("the first time to pull message, so fix offset from broker, offset maybe changed", map[string]interface{}{
 						rlog.LogKeyPullRequest:      request.String(),
@@ -675,7 +693,7 @@ func (pc *pushConsumer) pullMessage(request *PullRequest) {
 		)
 
 		if pc.model == Clustering {
-			commitOffsetValue = pc.storage.read(request.mq, _ReadFromMemory)
+			commitOffsetValue, _ = pc.storage.readWithException(request.mq, _ReadFromMemory)
 			if commitOffsetValue > 0 {
 				commitOffsetEnable = true
 			}
@@ -795,7 +813,7 @@ func (pc *pushConsumer) correctTagsOffset(pr *PullRequest) {
 func (pc *pushConsumer) sendMessageBack(brokerName string, msg *primitive.MessageExt, delayLevel int) bool {
 	var brokerAddr string
 	if len(brokerName) != 0 {
-		brokerAddr = pc.defaultConsumer.namesrv.FindBrokerAddrByName(brokerName)
+		brokerAddr = pc.defaultConsumer.client.GetNameSrv().FindBrokerAddrByName(brokerName)
 	} else {
 		brokerAddr = msg.StoreHost
 	}
diff --git a/examples/consumer/rpc/main.go b/examples/consumer/rpc/main.go
new file mode 100644
index 0000000..c2bf07c
--- /dev/null
+++ b/examples/consumer/rpc/main.go
@@ -0,0 +1,89 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/apache/rocketmq-client-go/v2/consumer"
+	"github.com/apache/rocketmq-client-go/v2/internal"
+	"github.com/apache/rocketmq-client-go/v2/primitive"
+	"github.com/apache/rocketmq-client-go/v2/producer"
+)
+
+const (
+	producerGroup = "please_rename_unique_group_name"
+	consumerGroup = "please_rename_unique_group_name"
+	topic         = "RequestTopic"
+)
+
+func main() {
+	// create a producer to send reply message
+	replyProducer, err := producer.NewDefaultProducer(
+		producer.WithGroupName(producerGroup),
+		producer.WithNsResolver(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
+	)
+	if err != nil {
+		fmt.Printf("error: %s\n", err)
+		return
+	}
+	err = replyProducer.Start()
+	if err != nil {
+		fmt.Printf("error: %s\n", err)
+		return
+	}
+
+	// create consumer
+	c, err := consumer.NewPushConsumer(
+		consumer.WithGroupName(consumerGroup),
+		consumer.WithConsumeFromWhere(consumer.ConsumeFromLastOffset),
+		consumer.WithPullInterval(0),
+		consumer.WithNsResolver(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
+	)
+
+	err = c.Subscribe(topic, consumer.MessageSelector{
+		Type: consumer.TAG, Expression: "*"}, func(ctx context.Context, msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
+
+		fmt.Printf("subscribe callback: %v \n", msgs)
+		for _, msg := range msgs {
+			fmt.Printf("handle message: %s", msg.String())
+			// why sleep here ? because if not sleep, consumer reply message, but producer not register to broker, it will
+			// make the broker can't find the reply channel.
+			// only the go client has this problem.
+			time.Sleep(time.Millisecond * 1000)
+			fmt.Println("consumer sleep over, start reply")
+
+			replyContent := []byte("reply message contents.")
+			replyMessage, err := internal.CreateReplyMessage(msg, replyContent)
+			if err != nil {
+				fmt.Printf("create reply message err:%v\n", err)
+				continue
+			}
+
+			replyTo := internal.GetReplyToClient(msg)
+			replyResult, err := replyProducer.SendSync(context.Background(), replyMessage)
+			if err != nil {
+				fmt.Printf("send message error: %s\n", err)
+				continue
+			}
+			fmt.Printf("reply to %s , %s \n", replyTo, replyResult.String())
+		}
+		return consumer.ConsumeSuccess, nil
+	})
+	if err != nil {
+		fmt.Printf("error: %s\n", err)
+		return
+	}
+	err = c.Start()
+	if err != nil {
+		fmt.Printf("error: %s\n", err)
+		return
+	}
+	fmt.Printf("Consumer Started.\n")
+
+	time.Sleep(time.Hour)
+	err = c.Shutdown()
+	if err != nil {
+		fmt.Printf("shutdown Consumer error: %s", err.Error())
+	}
+}
diff --git a/examples/producer/async/main.go b/examples/producer/async/main.go
index 2a5182c..e225214 100644
--- a/examples/producer/async/main.go
+++ b/examples/producer/async/main.go
@@ -32,8 +32,7 @@ import (
 func main() {
 	p, _ := rocketmq.NewProducer(
 		producer.WithNsResolver(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
-		producer.WithRetry(2),
-		producer.WithQueueSelector(producer.NewManualQueueSelector()))
+		producer.WithRetry(2))
 
 	err := p.Start()
 	if err != nil {
diff --git a/examples/producer/rpc/async/main.go b/examples/producer/rpc/async/main.go
new file mode 100644
index 0000000..4d226e5
--- /dev/null
+++ b/examples/producer/rpc/async/main.go
@@ -0,0 +1,52 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"time"
+
+	"github.com/apache/rocketmq-client-go/v2"
+	"github.com/apache/rocketmq-client-go/v2/primitive"
+	"github.com/apache/rocketmq-client-go/v2/producer"
+)
+
+func main() {
+	p, _ := rocketmq.NewProducer(
+		producer.WithGroupName("please_rename_unique_group_name"),
+		producer.WithNsResolver(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
+		producer.WithRetry(2),
+	)
+	err := p.Start()
+	if err != nil {
+		fmt.Printf("start producer error: %s", err.Error())
+		os.Exit(1)
+	}
+
+	topic := "RequestTopic"
+	ttl := 5 * time.Second
+	msg := &primitive.Message{
+		Topic: topic,
+		Body:  []byte("Hello RPC RocketMQ Go Client!"),
+	}
+
+	now := time.Now()
+	f := func(ctx context.Context, responseMsg *primitive.Message, respErr error) {
+		if respErr != nil {
+			fmt.Printf("request to <%s> fail, err:%v \n", topic, respErr)
+			return
+		}
+		fmt.Printf("Requst to %s cost:%d ms responseMsg:%s\n", topic, time.Since(now)/time.Millisecond, responseMsg.String())
+	}
+	err = p.RequestAsync(context.Background(), ttl, f, msg)
+	if err != nil {
+		fmt.Printf("Request Async message error: %s\n", err)
+		return
+	}
+
+	time.Sleep(time.Minute)
+	err = p.Shutdown()
+	if err != nil {
+		fmt.Printf("shutdown producer error: %s", err.Error())
+	}
+}
diff --git a/examples/producer/rpc/sync/main.go b/examples/producer/rpc/sync/main.go
new file mode 100644
index 0000000..b50ab35
--- /dev/null
+++ b/examples/producer/rpc/sync/main.go
@@ -0,0 +1,45 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"time"
+
+	"github.com/apache/rocketmq-client-go/v2"
+	"github.com/apache/rocketmq-client-go/v2/primitive"
+	"github.com/apache/rocketmq-client-go/v2/producer"
+)
+
+func main() {
+	p, _ := rocketmq.NewProducer(
+		producer.WithGroupName("please_rename_unique_group_name"),
+		producer.WithNsResolver(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
+		producer.WithRetry(2),
+	)
+	err := p.Start()
+	if err != nil {
+		fmt.Printf("start producer error: %s", err.Error())
+		os.Exit(1)
+	}
+
+	topic := "RequestTopic"
+	ttl := 5 * time.Second
+	msg := &primitive.Message{
+		Topic: topic,
+		Body:  []byte("Hello RPC RocketMQ Go Client!"),
+	}
+
+	now := time.Now()
+	responseMsg, err := p.Request(context.Background(), ttl, msg)
+	if err != nil {
+		fmt.Printf("Request message error: %s\n", err)
+		return
+	}
+	fmt.Printf("Requst to %s cost:%d ms responseMsg:%s\n", topic, time.Since(now)/time.Millisecond, responseMsg.String())
+
+	err = p.Shutdown()
+	if err != nil {
+		fmt.Printf("shutdown producer error: %s", err.Error())
+	}
+}
diff --git a/go.mod b/go.mod
index 80de649..93174fc 100644
--- a/go.mod
+++ b/go.mod
@@ -3,15 +3,21 @@ module github.com/apache/rocketmq-client-go/v2
 go 1.13
 
 require (
+	github.com/BurntSushi/toml v1.1.0 // indirect
 	github.com/emirpasic/gods v1.12.0
 	github.com/golang/mock v1.3.1
-	github.com/json-iterator/go v1.1.9
+	github.com/json-iterator/go v1.1.12
+	github.com/patrickmn/go-cache v2.1.0+incompatible
 	github.com/pkg/errors v0.8.1
+	github.com/satori/go.uuid v1.2.0
 	github.com/sirupsen/logrus v1.4.1
 	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.8.1
+	github.com/tidwall/gjson v1.13.0
 	go.uber.org/atomic v1.5.1
+	gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c // indirect
+	gopkg.in/natefinch/lumberjack.v2 v2.0.0
+	gopkg.in/yaml.v2 v2.4.0 // indirect
 	stathat.com/c/consistent v1.0.0 // indirect
 )
diff --git a/go.sum b/go.sum
index 58bed76..b27a55a 100644
--- a/go.sum
+++ b/go.sum
@@ -1,3 +1,5 @@
+github.com/BurntSushi/toml v1.1.0 h1:ksErzDEI1khOiGPgpwuI7x2ebx/uXQNw7xJpn9Eq1+I=
+github.com/BurntSushi/toml v1.1.0/go.mod h1:CxXYINrC8qIiEnFrOxCa7Jy5BFHlXnUU2pbicEuybxQ=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
@@ -8,20 +10,29 @@ github.com/golang/mock v1.3.1/go.mod h1:sBzyDLLjw3U8JLTeZvSv8jJB+tU5PVekmnlKIyFU
 github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
 github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1 h1:EGx4pi6eqNxGaHF6qqu48+N2wcFQ5qg5FXgOdqsJ5d8=
 github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1/go.mod h1:wJfORRmW1u3UXTncJ5qlYoELFm8eSnnEO6hX4iZ3EWY=
-github.com/json-iterator/go v1.1.9 h1:9yzud/Ht36ygwatGx56VwCZtlI/2AD15T1X2sjSuGns=
-github.com/json-iterator/go v1.1.9/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4=
+github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM=
+github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo=
 github.com/jtolds/gls v4.20.0+incompatible h1:xdiiI2gbIgH/gLH7ADydsJ1uDOEzR8yvV7C0MuV77Wo=
 github.com/jtolds/gls v4.20.0+incompatible/go.mod h1:QJZ7F/aHp+rZTRtaJ1ow/lLfFfVYBRgL+9YlvaHOwJU=
 github.com/konsorten/go-windows-terminal-sequences v1.0.1 h1:mweAR1A6xJ3oS2pRaGiHgQ4OO8tzTaLawm8vnODuwDk=
 github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
+github.com/kr/pretty v0.2.1 h1:Fmg33tUaq4/8ym9TJN1x7sLJnHVwhP33CNkpYV/7rwI=
+github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
+github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
+github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE=
+github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
 github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421 h1:ZqeYNhU3OHLH3mGKHDcjJRFFRrJa6eAM5H+CtDdOsPc=
 github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
-github.com/modern-go/reflect2 v0.0.0-20180701023420-4b7aa43c6742 h1:Esafd1046DLDQ0W1YjYsBW+p8U2u7vzgW2SQVmlNazg=
-github.com/modern-go/reflect2 v0.0.0-20180701023420-4b7aa43c6742/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0=
+github.com/modern-go/reflect2 v1.0.2 h1:xBagoLtFs94CBntxluKeaWgTMpvLxC4ur3nMaC9Gz0M=
+github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk=
+github.com/patrickmn/go-cache v2.1.0+incompatible h1:HRMgzkcYKYpi3C8ajMPV8OFXaaRUnok+kx1WdO15EQc=
+github.com/patrickmn/go-cache v2.1.0+incompatible/go.mod h1:3Qf8kWWT7OJRJbdiICTKqZju1ZixQ/KpMGzzAfe6+WQ=
 github.com/pkg/errors v0.8.1 h1:iURUrRGxPUNPdy5/HRSm+Yj6okJ6UtLINN0Q9M4+h3I=
 github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/satori/go.uuid v1.2.0 h1:0uYX9dsZ2yD7q2RtLRtPSdGDWzjeM3TbMJP9utgA0ww=
+github.com/satori/go.uuid v1.2.0/go.mod h1:dA0hQrYB0VpLJoorglMZABFdXlWrHn1NEOzdhQKdks0=
 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=
@@ -35,12 +46,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.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=
+github.com/tidwall/gjson v1.13.0 h1:3TFY9yxOQShrvmjdM76K+jc66zJeT6D3/VFFYCGQf7M=
+github.com/tidwall/gjson v1.13.0/go.mod h1:/wbyibRr2FHMks5tjHJ5F8dMZh3AcwJEMf5vlfC0lxk=
+github.com/tidwall/match v1.1.1 h1:+Ho715JplO36QYgwN9PGYNhgZvoUSc9X2c80KVTi+GA=
+github.com/tidwall/match v1.1.1/go.mod h1:eRSPERbgtNPcGhD8UCthc6PmLEQXEWd3PRB5JTxsfmM=
+github.com/tidwall/pretty v1.2.0 h1:RWIZEg2iJ8/g6fDDYzMpobmaoGh5OLl4AXtGUGPcqCs=
+github.com/tidwall/pretty v1.2.0/go.mod h1:ITEVvHYasfjBbM0u2Pg8T2nJnzm8xPwvNhhsoaGGjNU=
 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=
@@ -59,5 +70,12 @@ golang.org/x/tools v0.0.0-20190425150028-36563e24a262/go.mod h1:RgjU9mgBXZiqYHBn
 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=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
+gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
+gopkg.in/natefinch/lumberjack.v2 v2.0.0 h1:1Lc07Kr7qY4U2YPouBjpCLxpiyxIVoxqXgkXLknAOE8=
+gopkg.in/natefinch/lumberjack.v2 v2.0.0/go.mod h1:l0ndWWf7gzL7RNwBG7wST/UCcT4T24xpD6X8LsfU/+k=
+gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY=
+gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ=
 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 35a88a4..e8c4803 100644
--- a/internal/client.go
+++ b/internal/client.go
@@ -24,6 +24,7 @@ import (
 	errors2 "github.com/apache/rocketmq-client-go/v2/errors"
 	"net"
 	"os"
+	"sort"
 	"strconv"
 	"strings"
 	"sync"
@@ -84,7 +85,7 @@ type InnerConsumer interface {
 	SubscriptionDataList() []*SubscriptionData
 	Rebalance()
 	IsUnitMode() bool
-	GetConsumerRunningInfo() *ConsumerRunningInfo
+	GetConsumerRunningInfo(stack bool) *ConsumerRunningInfo
 	ConsumeMessageDirectly(msg *primitive.MessageExt, brokerName string) *ConsumeMessageDirectlyResult
 	GetcType() string
 	GetModel() string
@@ -94,32 +95,34 @@ type InnerConsumer interface {
 
 func DefaultClientOptions() ClientOptions {
 	opts := ClientOptions{
-		InstanceName: "DEFAULT",
-		RetryTimes:   3,
-		ClientIP:     utils.LocalIP,
+		InstanceName:         "DEFAULT",
+		RetryTimes:           3,
+		ClientIP:             utils.LocalIP,
+		RemotingClientConfig: &remote.DefaultRemotingClientConfig,
 	}
 	return opts
 }
 
 type ClientOptions struct {
-	GroupName         string
-	NameServerAddrs   primitive.NamesrvAddr
-	Namesrv           *namesrvs
-	ClientIP          string
-	InstanceName      string
-	UnitMode          bool
-	UnitName          string
-	VIPChannelEnabled bool
-	RetryTimes        int
-	Interceptors      []primitive.Interceptor
-	Credentials       primitive.Credentials
-	Namespace         string
-	Resolver          primitive.NsResolver
+	GroupName            string
+	NameServerAddrs      primitive.NamesrvAddr
+	Namesrv              Namesrvs
+	ClientIP             string
+	InstanceName         string
+	UnitMode             bool
+	UnitName             string
+	VIPChannelEnabled    bool
+	RetryTimes           int
+	Interceptors         []primitive.Interceptor
+	Credentials          primitive.Credentials
+	Namespace            string
+	Resolver             primitive.NsResolver
+	RemotingClientConfig *remote.RemotingClientConfig
 }
 
 func (opt *ClientOptions) ChangeInstanceNameToPID() {
 	if opt.InstanceName == "DEFAULT" {
-		opt.InstanceName = strconv.Itoa(os.Getpid())
+		opt.InstanceName = fmt.Sprintf("%d#%d", os.Getpid(), time.Now().UnixNano())
 	}
 }
 
@@ -136,7 +139,7 @@ type RMQClient interface {
 
 	ClientID() string
 
-	RegisterProducer(group string, producer InnerProducer)
+	RegisterProducer(group string, producer InnerProducer) error
 	UnregisterProducer(group string)
 	InvokeSync(ctx context.Context, addr string, request *remote.RemotingCommand,
 		timeoutMillis time.Duration) (*remote.RemotingCommand, error)
@@ -155,6 +158,8 @@ type RMQClient interface {
 	PullMessage(ctx context.Context, brokerAddrs string, request *PullMessageRequestHeader) (*primitive.PullResult, error)
 	RebalanceImmediately()
 	UpdatePublishInfo(topic string, data *TopicRouteData, changed bool)
+
+	GetNameSrv() Namesrvs
 }
 
 var _ RMQClient = new(rmqClient)
@@ -172,25 +177,48 @@ type rmqClient struct {
 	hbMutex      sync.Mutex
 	close        bool
 	rbMutex      sync.Mutex
-	namesrvs     *namesrvs
 	done         chan struct{}
 	shutdownOnce sync.Once
 }
 
+func (c *rmqClient) GetNameSrv() Namesrvs {
+	return c.option.Namesrv
+}
+
 var clientMap sync.Map
 
 func GetOrNewRocketMQClient(option ClientOptions, callbackCh chan interface{}) RMQClient {
 	client := &rmqClient{
 		option:       option,
-		remoteClient: remote.NewRemotingClient(),
-		namesrvs:     option.Namesrv,
+		remoteClient: remote.NewRemotingClient(option.RemotingClientConfig),
 		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 {
+
+	if loaded {
+		// compare namesrv address
+		client = actual.(*rmqClient)
+		now := option.Namesrv.(*namesrvs).resolver.Resolve()
+		old := client.GetNameSrv().(*namesrvs).resolver.Resolve()
+		if len(now) != len(old) {
+			rlog.Error("different namesrv option in the same instance", map[string]interface{}{
+				"NewNameSrv":    now,
+				"BeforeNameSrv": old,
+			})
+			return nil
+		}
+		sort.Strings(now)
+		sort.Strings(old)
+		for i := 0; i < len(now); i++ {
+			if now[i] != old[i] {
+				rlog.Error("different namesrv option in the same instance", map[string]interface{}{
+					"NewNameSrv":    now,
+					"BeforeNameSrv": old,
+				})
+				return nil
+			}
+		}
+	} else {
 		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{}{
 				rlog.LogKeyConsumerGroup: req.ExtFields["consumerGroup"],
@@ -242,7 +270,7 @@ func GetOrNewRocketMQClient(option ClientOptions, callbackCh chan interface{}) R
 				cli, ok := val.(*rmqClient)
 				var runningInfo *ConsumerRunningInfo
 				if ok {
-					runningInfo = cli.getConsumerRunningInfo(header.consumerGroup)
+					runningInfo = cli.getConsumerRunningInfo(header.consumerGroup, header.jstackEnable)
 				}
 				if runningInfo != nil {
 					res.Code = ResSuccess
@@ -305,8 +333,48 @@ func GetOrNewRocketMQClient(option ClientOptions, callbackCh chan interface{}) R
 			client.resetOffset(header.topic, header.group, body.OffsetTable)
 			return nil
 		})
+
+		client.remoteClient.RegisterRequestFunc(ReqPushReplyMessageToClient, func(req *remote.RemotingCommand, addr net.Addr) *remote.RemotingCommand {
+			receiveTime := time.Now().UnixNano() / int64(time.Millisecond)
+			rlog.Info("receive push reply to client request...", map[string]interface{}{
+				rlog.LogKeyBroker:        addr.String(),
+				rlog.LogKeyTopic:         req.ExtFields["topic"],
+				rlog.LogKeyConsumerGroup: req.ExtFields["group"],
+				rlog.LogKeyTimeStamp:     req.ExtFields["timestamp"],
+			})
+
+			header := new(ReplyMessageRequestHeader)
+			header.Decode(req.ExtFields)
+
+			var msgExt primitive.MessageExt
+			msgExt.Topic = header.topic
+			msgExt.Queue = &primitive.MessageQueue{
+				QueueId: header.queueId,
+				Topic:   header.topic,
+			}
+			msgExt.StoreTimestamp = header.storeTimestamp
+			msgExt.BornHost = header.bornHost
+			msgExt.StoreHost = header.storeHost
+
+			body := req.Body
+			if (header.sysFlag & primitive.FlagCompressed) == primitive.FlagCompressed {
+				body = utils.UnCompress(req.Body)
+			}
+			msgExt.Body = body
+			msgExt.Flag = header.flag
+			msgExt.UnmarshalProperties([]byte(header.properties))
+			msgExt.WithProperty(primitive.PropertyReplyMessageArriveTime, strconv.FormatInt(receiveTime, 10))
+			msgExt.BornTimestamp = header.bornTimestamp
+			msgExt.ReconsumeTimes = header.reconsumeTimes
+
+			client.getReplyMessageRequest(&msgExt, header.bornHost)
+
+			res := remote.NewRemotingCommand(ResError, nil, nil)
+			res.Code = ResSuccess
+			return res
+		})
 	}
-	return actual.(*rmqClient)
+	return client
 }
 
 func (c *rmqClient) Start() {
@@ -318,7 +386,7 @@ func (c *rmqClient) Start() {
 		}
 		go primitive.WithRecover(func() {
 			op := func() {
-				c.namesrvs.UpdateNameServerAddress()
+				c.GetNameSrv().UpdateNameServerAddress()
 			}
 			time.Sleep(10 * time.Second)
 			op()
@@ -364,7 +432,7 @@ func (c *rmqClient) Start() {
 
 		go primitive.WithRecover(func() {
 			op := func() {
-				c.namesrvs.cleanOfflineBroker()
+				c.GetNameSrv().cleanOfflineBroker()
 				c.SendHeartbeatToAllBrokerWithLock()
 			}
 
@@ -529,7 +597,7 @@ func (c *rmqClient) SendHeartbeatToAllBrokerWithLock() {
 		rlog.Info("sending heartbeat, but no producer and no consumer", nil)
 		return
 	}
-	c.namesrvs.brokerAddressesMap.Range(func(key, value interface{}) bool {
+	c.GetNameSrv().(*namesrvs).brokerAddressesMap.Range(func(key, value interface{}) bool {
 		brokerName := key.(string)
 		data := value.(*BrokerData)
 		for id, addr := range data.BrokerAddresses {
@@ -559,7 +627,7 @@ func (c *rmqClient) SendHeartbeatToAllBrokerWithLock() {
 			}
 			cancel()
 			if response.Code == ResSuccess {
-				c.namesrvs.AddBrokerVersion(brokerName, addr, int32(response.Version))
+				c.GetNameSrv().(*namesrvs).AddBrokerVersion(brokerName, addr, int32(response.Version))
 				rlog.Debug("send heart beat to broker success", map[string]interface{}{
 					"brokerName": brokerName,
 					"brokerId":   id,
@@ -589,7 +657,7 @@ func (c *rmqClient) UpdateTopicRouteInfo() {
 		return true
 	})
 	for topic := range publishTopicSet {
-		data, changed, _ := c.namesrvs.UpdateTopicRouteInfo(topic)
+		data, changed, _ := c.GetNameSrv().UpdateTopicRouteInfo(topic)
 		c.UpdatePublishInfo(topic, data, changed)
 	}
 
@@ -604,7 +672,7 @@ func (c *rmqClient) UpdateTopicRouteInfo() {
 	})
 
 	for topic := range subscribedTopicSet {
-		data, changed, _ := c.namesrvs.UpdateTopicRouteInfo(topic)
+		data, changed, _ := c.GetNameSrv().UpdateTopicRouteInfo(topic)
 		c.updateSubscribeInfo(topic, data, changed)
 	}
 }
@@ -730,8 +798,12 @@ func (c *rmqClient) UnregisterConsumer(group string) {
 	c.consumerMap.Delete(group)
 }
 
-func (c *rmqClient) RegisterProducer(group string, producer InnerProducer) {
-	c.producerMap.Store(group, producer)
+func (c *rmqClient) RegisterProducer(group string, producer InnerProducer) error {
+	_, loaded := c.producerMap.LoadOrStore(group, producer)
+	if loaded {
+		return fmt.Errorf("the producer group \"%s\" has been created, specify another one", c.option.GroupName)
+	}
+	return nil
 }
 
 func (c *rmqClient) UnregisterProducer(group string) {
@@ -760,7 +832,7 @@ func (c *rmqClient) UpdatePublishInfo(topic string, data *TopicRouteData, change
 			updated = p.IsPublishTopicNeedUpdate(topic)
 		}
 		if updated {
-			publishInfo := c.namesrvs.routeData2PublishInfo(topic, data)
+			publishInfo := c.GetNameSrv().(*namesrvs).routeData2PublishInfo(topic, data)
 			publishInfo.HaveTopicRouterInfo = true
 			p.UpdateTopicPublishInfo(topic, publishInfo)
 		}
@@ -808,18 +880,30 @@ func (c *rmqClient) resetOffset(topic string, group string, offsetTable map[prim
 	consumer.(InnerConsumer).ResetOffset(topic, offsetTable)
 }
 
-func (c *rmqClient) getConsumerRunningInfo(group string) *ConsumerRunningInfo {
+func (c *rmqClient) getConsumerRunningInfo(group string, stack bool) *ConsumerRunningInfo {
 	consumer, exist := c.consumerMap.Load(group)
 	if !exist {
 		return nil
 	}
-	info := consumer.(InnerConsumer).GetConsumerRunningInfo()
+	info := consumer.(InnerConsumer).GetConsumerRunningInfo(stack)
 	if info != nil {
 		info.Properties[PropClientVersion] = clientVersion
 	}
 	return info
 }
 
+func (c *rmqClient) getReplyMessageRequest(msg *primitive.MessageExt, bornHost string) {
+	correlationId := msg.GetProperty(primitive.PropertyCorrelationID)
+	if err := RequestResponseFutureMap.SetResponseToRequestResponseFuture(correlationId, &msg.Message); err != nil {
+		rlog.Warning("receive reply message, but not matched any request", map[string]interface{}{
+			"CorrelationId": correlationId,
+			"ReplyHost":     bornHost,
+		})
+		return
+	}
+	RequestResponseFutureMap.RemoveRequestResponseFuture(correlationId)
+}
+
 func (c *rmqClient) consumeMessageDirectly(msg *primitive.MessageExt, group string, brokerName string) *ConsumeMessageDirectlyResult {
 	consumer, exist := c.consumerMap.Load(group)
 	if !exist {
diff --git a/internal/constants.go b/internal/constants.go
index e2e911f..1f15a7e 100644
--- a/internal/constants.go
+++ b/internal/constants.go
@@ -22,8 +22,14 @@ const (
 	DefaultConsumerGroup     = "DEFAULT_CONSUMER"
 	ClientInnerProducerGroup = "CLIENT_INNER_PRODUCER"
 	SystemTopicPrefix        = "rmq_sys_"
+	ReplyMessageFlag         = "reply"
+	ReplyTopicPostfix        = "REPLY_TOPIC"
 )
 
+func GetReplyTopic(clusterName string) string {
+	return clusterName + "_" + ReplyTopicPostfix
+}
+
 func GetRetryTopic(group string) string {
 	return RetryGroupTopicPrefix + group
 }
diff --git a/internal/message_util.go b/internal/message_util.go
new file mode 100644
index 0000000..882b913
--- /dev/null
+++ b/internal/message_util.go
@@ -0,0 +1,39 @@
+package internal
+
+import (
+	"errors"
+	"fmt"
+
+	"github.com/apache/rocketmq-client-go/v2/primitive"
+)
+
+// CreateReplyMessage build reply message from the request message
+func CreateReplyMessage(requestMessage *primitive.MessageExt, body []byte) (*primitive.Message, error) {
+	if requestMessage == nil {
+		return nil, errors.New("create reply message fail, requestMessage cannot be null")
+	}
+
+	cluster := requestMessage.GetProperty(primitive.PropertyCluster)
+	replyTo := requestMessage.GetProperty(primitive.PropertyMessageReplyToClient)
+	correlationId := requestMessage.GetProperty(primitive.PropertyCorrelationID)
+	ttl := requestMessage.GetProperty(primitive.PropertyMessageTTL)
+
+	if cluster == "" {
+		return nil, fmt.Errorf("create reply message fail, requestMessage error, property[\"%s\"] is null", cluster)
+	}
+
+	var replayMessage primitive.Message
+
+	replayMessage.UnmarshalProperties(body)
+	replayMessage.Topic = GetReplyTopic(cluster)
+	replayMessage.WithProperty(primitive.PropertyMsgType, ReplyMessageFlag)
+	replayMessage.WithProperty(primitive.PropertyCorrelationID, correlationId)
+	replayMessage.WithProperty(primitive.PropertyMessageReplyToClient, replyTo)
+	replayMessage.WithProperty(primitive.PropertyMessageTTL, ttl)
+
+	return &replayMessage, nil
+}
+
+func GetReplyToClient(msg *primitive.MessageExt) string {
+	return msg.GetProperty(primitive.PropertyMessageReplyToClient)
+}
diff --git a/internal/mock_client.go b/internal/mock_client.go
index ab34ac1..c975038 100644
--- a/internal/mock_client.go
+++ b/internal/mock_client.go
@@ -208,6 +208,15 @@ func (mr *MockInnerConsumerMockRecorder) GetConsumerRunningInfo() *gomock.Call {
 type MockRMQClient struct {
 	ctrl     *gomock.Controller
 	recorder *MockRMQClientMockRecorder
+	Namesrv  *MockNamesrvs
+}
+
+func (m *MockRMQClient) GetNameSrv() Namesrvs {
+	return m.Namesrv
+}
+
+func (m *MockRMQClient) SetNameSrv(mockNamesrvs *MockNamesrvs) {
+	m.Namesrv = mockNamesrvs
 }
 
 // MockRMQClientMockRecorder is the mock recorder for MockRMQClient
@@ -260,8 +269,10 @@ func (mr *MockRMQClientMockRecorder) ClientID() *gomock.Call {
 }
 
 // RegisterProducer mocks base method
-func (m *MockRMQClient) RegisterProducer(group string, producer InnerProducer) {
-	m.ctrl.Call(m, "RegisterProducer", group, producer)
+func (m *MockRMQClient) RegisterProducer(group string, producer InnerProducer) error {
+	ret := m.ctrl.Call(m, "RegisterProducer", group, producer)
+	ret0, _ := ret[0].(error)
+	return ret0
 }
 
 // RegisterProducer indicates an expected call of RegisterProducer
diff --git a/internal/mock_namesrv.go b/internal/mock_namesrv.go
index f87d174..7ce6f97 100644
--- a/internal/mock_namesrv.go
+++ b/internal/mock_namesrv.go
@@ -33,6 +33,10 @@ type MockNamesrvs struct {
 	recorder *MockNamesrvsMockRecorder
 }
 
+func (m *MockNamesrvs) UpdateTopicRouteInfoWithDefault(topic string, defaultTopic string, defaultQueueNum int) (*TopicRouteData, bool, error) {
+	return m.UpdateTopicRouteInfo(topic)
+}
+
 // MockNamesrvsMockRecorder is the mock recorder for MockNamesrvs
 type MockNamesrvsMockRecorder struct {
 	mock *MockNamesrvs
diff --git a/internal/model.go b/internal/model.go
index 7a011d7..c248bb3 100644
--- a/internal/model.go
+++ b/internal/model.go
@@ -149,6 +149,7 @@ type ConsumerRunningInfo struct {
 	SubscriptionData map[*SubscriptionData]bool
 	MQTable          map[primitive.MessageQueue]ProcessQueueInfo
 	StatusTable      map[string]ConsumeStatus
+	JStack           string // just follow java request param name, but pass golang stack here.
 }
 
 func (info ConsumerRunningInfo) Encode() ([]byte, error) {
@@ -251,7 +252,11 @@ func (info ConsumerRunningInfo) Encode() ([]byte, error) {
 		tableJson = fmt.Sprintf("%s,%s:%s", tableJson, string(dataK), string(dataV))
 	}
 	tableJson = strings.TrimLeft(tableJson, ",")
-	jsonData = fmt.Sprintf("%s,\"%s\":%s}", jsonData, "mqTable", fmt.Sprintf("{%s}", tableJson))
+
+	jsonData = fmt.Sprintf("%s,\"%s\":%s, \"%s\":\"%s\" }",
+		jsonData, "mqTable", fmt.Sprintf("{%s}", tableJson),
+		"jstack", info.JStack)
+
 	return []byte(jsonData), nil
 }
 
diff --git a/internal/namesrv.go b/internal/namesrv.go
index 7776651..c347998 100644
--- a/internal/namesrv.go
+++ b/internal/namesrv.go
@@ -50,6 +50,8 @@ type Namesrvs interface {
 
 	UpdateTopicRouteInfo(topic string) (routeData *TopicRouteData, changed bool, err error)
 
+	UpdateTopicRouteInfoWithDefault(topic string, defaultTopic string, defaultQueueNum int) (*TopicRouteData, bool, error)
+
 	FetchPublishMessageQueues(topic string) ([]*primitive.MessageQueue, error)
 
 	FindBrokerAddrByTopic(topic string) string
@@ -95,22 +97,18 @@ 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)
+	actual, ok := clientMap.Load(clientId)
 	if !ok {
 		return nil, fmt.Errorf("the namesrv in instanceName [%s] not found", clientId)
 	}
-	return actual.(*namesrvs), nil
+	return actual.(*rmqClient).GetNameSrv().(*namesrvs), nil
 }
-func NewNamesrv(resolver primitive.NsResolver) (*namesrvs, error) {
+
+// NewNamesrv init Namesrv from namesrv addr string.
+// addr primitive.NamesrvAddr
+func NewNamesrv(resolver primitive.NsResolver, config *remote.RemotingClientConfig) (*namesrvs, error) {
 	addr := resolver.Resolve()
 	if len(addr) == 0 {
 		return nil, errors.New("no name server addr found with resolver: " + resolver.Description())
@@ -119,7 +117,7 @@ func NewNamesrv(resolver primitive.NsResolver) (*namesrvs, error) {
 	if err := primitive.NamesrvAddr(addr).Check(); err != nil {
 		return nil, err
 	}
-	nameSrvClient := remote.NewRemotingClient()
+	nameSrvClient := remote.NewRemotingClient(config)
 	return &namesrvs{
 		srvs:             addr,
 		lock:             new(sync.Mutex),
diff --git a/internal/remote/remote_client.go b/internal/remote/remote_client.go
index 33b4633..4013689 100644
--- a/internal/remote/remote_client.go
+++ b/internal/remote/remote_client.go
@@ -24,6 +24,7 @@ import (
 	"io"
 	"net"
 	"sync"
+	"time"
 
 	"github.com/apache/rocketmq-client-go/v2/primitive"
 
@@ -33,7 +34,10 @@ import (
 type ClientRequestFunc func(*RemotingCommand, net.Addr) *RemotingCommand
 
 type TcpOption struct {
-	// TODO
+	KeepAliveDuration time.Duration
+	ConnectionTimeout time.Duration
+	ReadTimeout       time.Duration
+	WriteTimeout      time.Duration
 }
 
 //go:generate mockgen -source remote_client.go -destination mock_remote_client.go -self_package github.com/apache/rocketmq-client-go/v2/internal/remote  --package remote RemotingClient
@@ -51,15 +55,33 @@ var _ RemotingClient = &remotingClient{}
 type remotingClient struct {
 	responseTable    sync.Map
 	connectionTable  sync.Map
-	option           TcpOption
+	config           *RemotingClientConfig
 	processors       map[int16]ClientRequestFunc
 	connectionLocker sync.Mutex
 	interceptor      primitive.Interceptor
 }
 
-func NewRemotingClient() *remotingClient {
+type RemotingClientConfig struct {
+	TcpOption
+}
+
+var DefaultRemotingClientConfig = RemotingClientConfig{defaultTcpOption}
+
+var defaultTcpOption = TcpOption{
+	KeepAliveDuration: 0, // default 15s in golang
+	ConnectionTimeout: time.Second * 15,
+	ReadTimeout:       time.Second * 120,
+	WriteTimeout:      time.Second * 120,
+}
+
+func NewRemotingClient(config *RemotingClientConfig) *remotingClient {
+	if config == nil {
+		config = &DefaultRemotingClientConfig
+	}
+
 	return &remotingClient{
 		processors: make(map[int16]ClientRequestFunc),
+		config:     config,
 	}
 }
 
@@ -73,13 +95,17 @@ func (c *remotingClient) InvokeSync(ctx context.Context, addr string, request *R
 	if err != nil {
 		return nil, err
 	}
+
 	resp := NewResponseFuture(ctx, request.Opaque, nil)
+
 	c.responseTable.Store(resp.Opaque, resp)
 	defer c.responseTable.Delete(request.Opaque)
+
 	err = c.sendRequest(conn, request)
 	if err != nil {
 		return nil, err
 	}
+
 	return resp.waitResponse()
 }
 
@@ -89,15 +115,21 @@ func (c *remotingClient) InvokeAsync(ctx context.Context, addr string, request *
 	if err != nil {
 		return err
 	}
+
 	resp := NewResponseFuture(ctx, request.Opaque, callback)
 	c.responseTable.Store(resp.Opaque, resp)
+
 	err = c.sendRequest(conn, request)
 	if err != nil {
+		c.responseTable.Delete(request.Opaque)
 		return err
 	}
+
 	go primitive.WithRecover(func() {
 		c.receiveAsync(resp)
+		c.responseTable.Delete(request.Opaque)
 	})
+
 	return nil
 }
 
@@ -124,7 +156,7 @@ func (c *remotingClient) connect(ctx context.Context, addr string) (*tcpConnWrap
 	if ok {
 		return conn.(*tcpConnWrapper), nil
 	}
-	tcpConn, err := initConn(ctx, addr)
+	tcpConn, err := initConn(ctx, addr, c.config)
 	if err != nil {
 		return nil, err
 	}
@@ -155,6 +187,11 @@ func (c *remotingClient) receiveResponse(r *tcpConnWrapper) {
 			break
 		}
 
+		err = r.Conn.SetReadDeadline(time.Now().Add(c.config.ReadTimeout))
+		if err != nil {
+			continue
+		}
+
 		_, err = io.ReadFull(r, *header)
 		if err != nil {
 			continue
@@ -274,11 +311,29 @@ func (c *remotingClient) sendRequest(conn *tcpConnWrapper, request *RemotingComm
 func (c *remotingClient) doRequest(conn *tcpConnWrapper, request *RemotingCommand) error {
 	conn.Lock()
 	defer conn.Unlock()
-	err := request.WriteTo(conn)
+
+	err := conn.Conn.SetWriteDeadline(time.Now().Add(c.config.WriteTimeout))
 	if err != nil {
+		rlog.Error("conn error, close connection", map[string]interface{}{
+			rlog.LogKeyUnderlayError: err,
+		})
+
 		c.closeConnection(conn)
+		conn.destroy()
 		return err
 	}
+
+	err = request.WriteTo(conn)
+	if err != nil {
+		rlog.Error("conn error, close connection", map[string]interface{}{
+			rlog.LogKeyUnderlayError: err,
+		})
+
+		c.closeConnection(conn)
+		conn.destroy()
+		return err
+	}
+
 	return nil
 }
 
diff --git a/internal/remote/tcp_conn.go b/internal/remote/tcp_conn.go
index ae340c6..93ed837 100644
--- a/internal/remote/tcp_conn.go
+++ b/internal/remote/tcp_conn.go
@@ -20,6 +20,7 @@ import (
 	"context"
 	"net"
 	"sync"
+	"time"
 
 	"go.uber.org/atomic"
 )
@@ -31,8 +32,12 @@ type tcpConnWrapper struct {
 	closed atomic.Bool
 }
 
-func initConn(ctx context.Context, addr string) (*tcpConnWrapper, error) {
+func initConn(ctx context.Context, addr string, config *RemotingClientConfig) (*tcpConnWrapper, error) {
 	var d net.Dialer
+
+	d.KeepAlive = config.KeepAliveDuration
+	d.Deadline = time.Now().Add(config.ConnectionTimeout)
+
 	conn, err := d.DialContext(ctx, "tcp", addr)
 	if err != nil {
 		return nil, err
diff --git a/internal/request.go b/internal/request.go
index 0e3d8e1..5ff04e3 100644
--- a/internal/request.go
+++ b/internal/request.go
@@ -51,6 +51,9 @@ const (
 	ReqResetConsumerOffset           = int16(220)
 	ReqGetConsumerRunningInfo        = int16(307)
 	ReqConsumeMessageDirectly        = int16(309)
+	ReqSendReplyMessage              = int16(324)
+	ReqSendReplyMessageV2            = int16(325)
+	ReqPushReplyMessageToClient      = int16(326)
 )
 
 type SendMessageRequestHeader struct {
@@ -305,12 +308,14 @@ func (request *GetRouteInfoRequestHeader) Encode() map[string]string {
 type GetConsumerRunningInfoHeader struct {
 	consumerGroup string
 	clientID      string
+	jstackEnable  bool
 }
 
 func (request *GetConsumerRunningInfoHeader) Encode() map[string]string {
 	maps := make(map[string]string)
 	maps["consumerGroup"] = request.consumerGroup
 	maps["clientId"] = request.clientID
+	maps["jstackEnable"] = strconv.FormatBool(request.jstackEnable)
 	return maps
 }
 
@@ -325,6 +330,11 @@ func (request *GetConsumerRunningInfoHeader) Decode(properties map[string]string
 	if v, existed := properties["clientId"]; existed {
 		request.clientID = v
 	}
+
+	if v, existed := properties["jstackEnable"]; existed {
+		parseBool, _ := strconv.ParseBool(v)
+		request.jstackEnable = parseBool
+	}
 }
 
 type QueryMessageRequestHeader struct {
@@ -478,3 +488,98 @@ func (request *ConsumeMessageDirectlyHeader) Decode(properties map[string]string
 		request.brokerName = v
 	}
 }
+
+type ReplyMessageRequestHeader struct {
+	producerGroup         string
+	topic                 string
+	defaultTopic          string
+	defaultTopicQueueNums int
+	queueId               int
+	sysFlag               int
+	bornTimestamp         int64
+	flag                  int32
+	properties            string
+	reconsumeTimes        int32
+	unitMode              bool
+	bornHost              string
+	storeHost             string
+	storeTimestamp        int64
+}
+
+func (request *ReplyMessageRequestHeader) Encode() map[string]string {
+	return map[string]string{
+		"producerGroup":         request.producerGroup,
+		"topic":                 request.topic,
+		"defaultTopic":          request.defaultTopic,
+		"defaultTopicQueueNums": strconv.Itoa(request.defaultTopicQueueNums),
+		"queueId":               strconv.Itoa(request.queueId),
+		"sysFlag":               strconv.Itoa(request.sysFlag),
+		"bornTimestamp":         strconv.FormatInt(request.bornTimestamp, 10),
+		"flag":                  fmt.Sprintf("%d", request.flag),
+		"properties":            request.properties,
+		"reconsumeTimes":        fmt.Sprintf("%d", request.reconsumeTimes),
+		"bornHost":              request.bornHost,
+		"storeHost":             request.storeHost,
+		"storeTimestamp":        strconv.FormatInt(request.storeTimestamp, 10),
+	}
+}
+
+func (request *ReplyMessageRequestHeader) Decode(properties map[string]string) {
+	if len(properties) == 0 {
+		return
+	}
+
+	if v, existed := properties["producerGroup"]; existed {
+		request.producerGroup = v
+	}
+
+	if v, existed := properties["topic"]; existed {
+		request.topic = v
+	}
+
+	if v, existed := properties["defaultTopic"]; existed {
+		request.defaultTopic = v
+	}
+
+	if v, existed := properties["defaultTopicQueueNums"]; existed {
+		request.defaultTopicQueueNums, _ = strconv.Atoi(v)
+	}
+
+	if v, existed := properties["queueId"]; existed {
+		request.queueId, _ = strconv.Atoi(v)
+	}
+
+	if v, existed := properties["sysFlag"]; existed {
+		request.sysFlag, _ = strconv.Atoi(v)
+	}
+
+	if v, existed := properties["bornTimestamp"]; existed {
+		request.bornTimestamp, _ = strconv.ParseInt(v, 10, 0)
+	}
+
+	if v, existed := properties["flag"]; existed {
+		tmpFlag, _ := strconv.ParseInt(v, 10, 32)
+		request.flag = int32(tmpFlag)
+	}
+
+	if v, existed := properties["properties"]; existed {
+		request.properties = v
+	}
+
+	if v, existed := properties["reconsumeTimes"]; existed {
+		tmpReconsumeTimes, _ := strconv.ParseInt(v, 10, 32)
+		request.reconsumeTimes = int32(tmpReconsumeTimes)
+	}
+
+	if v, existed := properties["bornHost"]; existed {
+		request.bornHost = v
+	}
+
+	if v, existed := properties["storeHost"]; existed {
+		request.storeHost = v
+	}
+
+	if v, existed := properties["storeTimestamp"]; existed {
+		request.storeTimestamp, _ = strconv.ParseInt(v, 10, 0)
+	}
+}
diff --git a/internal/request_response_future.go b/internal/request_response_future.go
new file mode 100644
index 0000000..ed688a4
--- /dev/null
+++ b/internal/request_response_future.go
@@ -0,0 +1,133 @@
+package internal
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/patrickmn/go-cache"
+	"github.com/pkg/errors"
+
+	"github.com/apache/rocketmq-client-go/v2/primitive"
+	"github.com/apache/rocketmq-client-go/v2/rlog"
+)
+
+var RequestResponseFutureMap = NewRequestResponseFutureMap()
+
+type requestResponseFutureCache struct {
+	cache *cache.Cache
+}
+
+func NewRequestResponseFutureMap() *requestResponseFutureCache {
+	tmpRrfCache := requestResponseFutureCache{
+		cache: cache.New(5*time.Minute, 10*time.Minute),
+	}
+
+	// OnEvicted delete the timeout RequestResponseFuture, trigger set the failure cause.
+	tmpRrfCache.cache.OnEvicted(func(s string, i interface{}) {
+		rrf, ok := i.(*RequestResponseFuture)
+		if !ok {
+			rlog.Error("convert i to RequestResponseFuture err", map[string]interface{}{
+				"correlationId": s,
+			})
+			return
+		}
+
+		if rrf.IsTimeout() {
+			rrf.CauseErr = fmt.Errorf("correlationId:%s request timeout, no reply message", s)
+		}
+		rrf.ExecuteRequestCallback()
+	})
+	return &tmpRrfCache
+}
+
+// SetRequestResponseFuture set rrf to map
+func (fm *requestResponseFutureCache) SetRequestResponseFuture(rrf *RequestResponseFuture) {
+	fm.cache.Set(rrf.CorrelationId, rrf, rrf.Timeout)
+}
+
+// SetResponseToRequestResponseFuture set reply to rrf
+func (fm *requestResponseFutureCache) SetResponseToRequestResponseFuture(correlationId string, reply *primitive.Message) error {
+	rrf, exist := fm.RequestResponseFuture(correlationId)
+	if !exist {
+		return errors.Wrapf(nil, "correlationId:%s not exist in map", correlationId)
+	}
+	rrf.PutResponseMessage(reply)
+	if rrf.RequestCallback != nil {
+		rrf.ExecuteRequestCallback()
+	}
+	return nil
+}
+
+// RequestResponseFuture get rrf from map by the CorrelationId
+func (fm *requestResponseFutureCache) RequestResponseFuture(correlationId string) (*RequestResponseFuture, bool) {
+	res, exists := fm.cache.Get(correlationId)
+	if exists {
+		return res.(*RequestResponseFuture), exists
+	}
+	return nil, exists
+}
+
+// RemoveRequestResponseFuture remove the rrf from map
+func (fm *requestResponseFutureCache) RemoveRequestResponseFuture(correlationId string) {
+	fm.cache.Delete(correlationId)
+}
+
+type RequestCallback func(ctx context.Context, msg *primitive.Message, err error)
+
+// RequestResponseFuture store the rpc request. When producer wait for the response, get RequestResponseFuture.
+type RequestResponseFuture struct {
+	CorrelationId   string
+	mtx             sync.RWMutex
+	ResponseMsg     *primitive.Message
+	Timeout         time.Duration
+	RequestCallback RequestCallback
+	SendRequestOk   bool
+	Done            chan struct{}
+	CauseErr        error
+	BeginTime       time.Time
+}
+
+func NewRequestResponseFuture(correlationId string, timeout time.Duration, callback RequestCallback) *RequestResponseFuture {
+	return &RequestResponseFuture{
+		CorrelationId:   correlationId,
+		Timeout:         timeout,
+		RequestCallback: callback,
+		Done:            make(chan struct{}),
+		BeginTime:       time.Now(),
+	}
+}
+
+func (rf *RequestResponseFuture) ExecuteRequestCallback() {
+	if rf.RequestCallback == nil {
+		return
+	}
+
+	rf.RequestCallback(context.Background(), rf.ResponseMsg, rf.CauseErr)
+}
+
+func (rf *RequestResponseFuture) WaitResponseMessage(reqMsg *primitive.Message) (*primitive.Message, error) {
+	select {
+	case <-time.After(rf.Timeout):
+		err := fmt.Errorf("send request message to %s OK, but wait reply message timeout %d ms", reqMsg.Topic, rf.Timeout/time.Millisecond)
+		rlog.Error(err.Error(), nil)
+		return nil, err
+	case <-rf.Done:
+		rf.mtx.RLock()
+		rf.mtx.RUnlock()
+		return rf.ResponseMsg, nil
+	}
+}
+
+func (rf *RequestResponseFuture) PutResponseMessage(message *primitive.Message) {
+	rf.mtx.Lock()
+	defer rf.mtx.Unlock()
+	rf.ResponseMsg = message
+	close(rf.Done)
+}
+
+func (rf *RequestResponseFuture) IsTimeout() bool {
+	diff := time.Since(rf.BeginTime)
+	return diff > rf.Timeout
+}
diff --git a/internal/route.go b/internal/route.go
index 9cfa398..a101a61 100644
--- a/internal/route.go
+++ b/internal/route.go
@@ -112,6 +112,14 @@ func (s *namesrvs) UpdateTopicRouteInfo(topic string) (*TopicRouteData, bool, er
 	return s.UpdateTopicRouteInfoWithDefault(topic, "", 0)
 }
 
+func (s *namesrvs) CheckTopicRouteHasTopic(topic string) bool {
+	_, err := s.queryTopicRouteInfoFromServer(topic)
+	if err != nil {
+		return false
+	}
+	return true
+}
+
 func (s *namesrvs) UpdateTopicRouteInfoWithDefault(topic string, defaultTopic string, defaultQueueNum int) (*TopicRouteData, bool, error) {
 	s.lockNamesrv.Lock()
 	defer s.lockNamesrv.Unlock()
@@ -165,7 +173,7 @@ func (s *namesrvs) UpdateTopicRouteInfoWithDefault(topic string, defaultTopic st
 					updated = p.IsPublishTopicNeedUpdate(topic)
 				}
 				if updated {
-					publishInfo := s.bundleClient.namesrvs.routeData2PublishInfo(topic, routeData)
+					publishInfo := s.bundleClient.GetNameSrv().(*namesrvs).routeData2PublishInfo(topic, routeData)
 					publishInfo.HaveTopicRouterInfo = true
 					p.UpdateTopicPublishInfo(topic, publishInfo)
 				}
diff --git a/internal/trace.go b/internal/trace.go
index cef1634..cefc7ba 100644
--- a/internal/trace.go
+++ b/internal/trace.go
@@ -256,9 +256,9 @@ func NewTraceDispatcher(traceCfg *primitive.TraceConfig) *traceDispatcher {
 	var srvs *namesrvs
 	var err error
 	if len(traceCfg.NamesrvAddrs) > 0 {
-		srvs, err = NewNamesrv(primitive.NewPassthroughResolver(traceCfg.NamesrvAddrs))
+		srvs, err = NewNamesrv(primitive.NewPassthroughResolver(traceCfg.NamesrvAddrs), nil)
 	} else {
-		srvs, err = NewNamesrv(traceCfg.Resolver)
+		srvs, err = NewNamesrv(traceCfg.Resolver, nil)
 	}
 
 	if err != nil {
@@ -276,6 +276,10 @@ func NewTraceDispatcher(traceCfg *primitive.TraceConfig) *traceDispatcher {
 	cliOp.Namesrv = srvs
 	cliOp.Credentials = traceCfg.Credentials
 	cli := GetOrNewRocketMQClient(cliOp, nil)
+	if cli == nil {
+		return nil
+	}
+	cliOp.Namesrv = cli.GetNameSrv()
 	return &traceDispatcher{
 		ctx:    ctx,
 		cancel: cancel,
diff --git a/primitive/message.go b/primitive/message.go
index b8e8f83..32adc2b 100644
--- a/primitive/message.go
+++ b/primitive/message.go
@@ -60,15 +60,22 @@ const (
 	PropertyCheckImmunityTimeInSeconds     = "CHECK_IMMUNITY_TIME_IN_SECONDS"
 	PropertyShardingKey                    = "SHARDING_KEY"
 	PropertyTransactionID                  = "__transactionId__"
+	PropertyCorrelationID                  = "CORRELATION_ID"
+	PropertyMessageReplyToClient           = "REPLY_TO_CLIENT"
+	PropertyMessageTTL                     = "TTL"
+	PropertyReplyMessageArriveTime         = "ARRIVE_TIME"
+	PropertyMsgType                        = "MSG_TYPE"
+	PropertyCluster                        = "CLUSTER"
 )
 
 type Message struct {
-	Topic         string
-	Body          []byte
-	Flag          int32
-	TransactionId string
-	Batch         bool
-	Compress      bool
+	Topic          string
+	Body           []byte
+	CompressedBody []byte
+	Flag           int32
+	TransactionId  string
+	Batch          bool
+	Compress       bool
 	// Queue is the queue that messages will be sent to. the value must be set if want to custom the queue of message,
 	// just ignore if not.
 	Queue *MessageQueue
diff --git a/producer/interceptor.go b/producer/interceptor.go
index 160deac..71eb8e7 100644
--- a/producer/interceptor.go
+++ b/producer/interceptor.go
@@ -22,6 +22,7 @@ package producer
 
 import (
 	"context"
+	"fmt"
 	"time"
 
 	"github.com/apache/rocketmq-client-go/v2/internal"
@@ -42,9 +43,14 @@ func WithTrace(traceCfg *primitive.TraceConfig) Option {
 
 func newTraceInterceptor(traceCfg *primitive.TraceConfig) primitive.Interceptor {
 	dispatcher := internal.NewTraceDispatcher(traceCfg)
-	dispatcher.Start()
+	if dispatcher != nil {
+		dispatcher.Start()
+	}
 
 	return func(ctx context.Context, req, reply interface{}, next primitive.Invoker) error {
+		if dispatcher == nil {
+			return fmt.Errorf("GetOrNewRocketMQClient faild")
+		}
 		beginT := time.Now()
 		err := next(ctx, req, reply)
 
diff --git a/producer/option.go b/producer/option.go
index 5839402..ae76511 100644
--- a/producer/option.go
+++ b/producer/option.go
@@ -35,7 +35,7 @@ func defaultProducerOptions() producerOptions {
 		CompressMsgBodyOverHowmuch: 4096,
 		CompressLevel:              5,
 	}
-	opts.ClientOptions.GroupName = "DEFAULT_CONSUMER"
+	opts.ClientOptions.GroupName = "DEFAULT_PRODUCER"
 	return opts
 }
 
diff --git a/producer/producer.go b/producer/producer.go
index 226eedb..5ebf5d0 100644
--- a/producer/producer.go
+++ b/producer/producer.go
@@ -21,14 +21,15 @@ import (
 	"bytes"
 	"context"
 	"fmt"
-	errors2 "github.com/apache/rocketmq-client-go/v2/errors"
 	"strconv"
 	"sync"
 	"sync/atomic"
 	"time"
 
 	"github.com/pkg/errors"
+	uuid "github.com/satori/go.uuid"
 
+	errors2 "github.com/apache/rocketmq-client-go/v2/errors"
 	"github.com/apache/rocketmq-client-go/v2/internal"
 	"github.com/apache/rocketmq-client-go/v2/internal/remote"
 	"github.com/apache/rocketmq-client-go/v2/internal/utils"
@@ -52,7 +53,7 @@ func NewDefaultProducer(opts ...Option) (*defaultProducer, error) {
 	for _, apply := range opts {
 		apply(&defaultOpts)
 	}
-	srvs, err := internal.NewNamesrv(defaultOpts.Resolver)
+	srvs, err := internal.NewNamesrv(defaultOpts.Resolver, defaultOpts.RemotingClientConfig)
 	if err != nil {
 		return nil, errors.Wrap(err, "new Namesrv failed.")
 	}
@@ -67,19 +68,25 @@ 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
+	if producer.client == nil {
+		return nil, fmt.Errorf("GetOrNewRocketMQClient faild")
 	}
+	defaultOpts.Namesrv = producer.client.GetNameSrv()
+
 	producer.interceptor = primitive.ChainInterceptors(producer.options.Interceptors...)
 
 	return producer, nil
 }
 
 func (p *defaultProducer) Start() error {
+	if p == nil || p.client == nil {
+		return fmt.Errorf("client instance is nil, can not start producer")
+	}
 	atomic.StoreInt32(&p.state, int32(internal.StateRunning))
-
-	p.client.RegisterProducer(p.group, p)
+	err := p.client.RegisterProducer(p.group, p)
+	if err != nil {
+		return err
+	}
 	p.client.Start()
 	return nil
 }
@@ -144,6 +151,111 @@ func MarshalMessageBatch(msgs ...*primitive.Message) []byte {
 	return buffer.Bytes()
 }
 
+func (p *defaultProducer) prepareSendRequest(msg *primitive.Message, ttl time.Duration) (string, error) {
+	correlationId := uuid.NewV4().String()
+	requestClientId := p.client.ClientID()
+	msg.WithProperty(primitive.PropertyCorrelationID, correlationId)
+	msg.WithProperty(primitive.PropertyMessageReplyToClient, requestClientId)
+	msg.WithProperty(primitive.PropertyMessageTTL, strconv.Itoa(int(ttl.Seconds())))
+
+	rlog.Debug("message info:", map[string]interface{}{
+		"clientId":      requestClientId,
+		"correlationId": correlationId,
+		"ttl":           ttl.Seconds(),
+	})
+
+	nameSrv, err := internal.GetNamesrv(requestClientId)
+	if err != nil {
+		return "", errors.Wrap(err, "GetNameServ err")
+	}
+
+	if !nameSrv.CheckTopicRouteHasTopic(msg.Topic) {
+		p.tryToFindTopicPublishInfo(msg.Topic)
+		p.client.SendHeartbeatToAllBrokerWithLock()
+	}
+	return correlationId, nil
+}
+
+// Request Send messages to consumer
+func (p *defaultProducer) Request(ctx context.Context, timeout time.Duration, msg *primitive.Message) (*primitive.Message, error) {
+	if err := p.checkMsg(msg); err != nil {
+		return nil, err
+	}
+
+	p.messagesWithNamespace(msg)
+	correlationId, err := p.prepareSendRequest(msg, timeout)
+	if err != nil {
+		return nil, err
+	}
+
+	requestResponseFuture := internal.NewRequestResponseFuture(correlationId, timeout, nil)
+	internal.RequestResponseFutureMap.SetRequestResponseFuture(requestResponseFuture)
+	defer internal.RequestResponseFutureMap.RemoveRequestResponseFuture(correlationId)
+
+	f := func(ctx context.Context, result *primitive.SendResult, err error) {
+		if err != nil {
+			requestResponseFuture.SendRequestOk = false
+			requestResponseFuture.ResponseMsg = nil
+			requestResponseFuture.CauseErr = err
+			return
+		}
+		requestResponseFuture.SendRequestOk = true
+	}
+
+	if p.interceptor != nil {
+		primitive.WithMethod(ctx, primitive.SendAsync)
+
+		return nil, p.interceptor(ctx, msg, nil, func(ctx context.Context, req, reply interface{}) error {
+			return p.sendAsync(ctx, msg, f)
+		})
+	}
+	if err := p.sendAsync(ctx, msg, f); err != nil {
+		return nil, errors.Wrap(err, "sendAsync error")
+	}
+
+	return requestResponseFuture.WaitResponseMessage(msg)
+}
+
+// RequestAsync  Async Send messages to consumer
+func (p *defaultProducer) RequestAsync(ctx context.Context, timeout time.Duration, callback internal.RequestCallback, msg *primitive.Message) error {
+	if err := p.checkMsg(msg); err != nil {
+		return err
+	}
+
+	p.messagesWithNamespace(msg)
+	correlationId, err := p.prepareSendRequest(msg, timeout)
+	if err != nil {
+		return err
+	}
+
+	requestResponseFuture := internal.NewRequestResponseFuture(correlationId, timeout, callback)
+	internal.RequestResponseFutureMap.SetRequestResponseFuture(requestResponseFuture)
+
+	f := func(ctx context.Context, result *primitive.SendResult, err error) {
+		if err != nil {
+			requestResponseFuture.SendRequestOk = false
+			requestResponseFuture.ResponseMsg = nil
+			requestResponseFuture.CauseErr = err
+			internal.RequestResponseFutureMap.RemoveRequestResponseFuture(correlationId)
+			return
+		}
+		requestResponseFuture.SendRequestOk = true
+	}
+
+	var resErr error
+	if p.interceptor != nil {
+		primitive.WithMethod(ctx, primitive.SendAsync)
+		resErr = p.interceptor(ctx, msg, nil, func(ctx context.Context, req, reply interface{}) error {
+			return p.sendAsync(ctx, msg, f)
+		})
+	}
+	resErr = p.sendAsync(ctx, msg, f)
+	if resErr != nil {
+		internal.RequestResponseFutureMap.RemoveRequestResponseFuture(correlationId)
+	}
+	return resErr
+}
+
 func (p *defaultProducer) SendSync(ctx context.Context, msgs ...*primitive.Message) (*primitive.SendResult, error) {
 	if err := p.checkMsg(msgs...); err != nil {
 		return nil, err
@@ -195,7 +307,7 @@ func (p *defaultProducer) sendSync(ctx context.Context, msg *primitive.Message,
 			continue
 		}
 
-		addr := p.options.Namesrv.FindBrokerAddrByName(mq.BrokerName)
+		addr := p.client.GetNameSrv().FindBrokerAddrByName(mq.BrokerName)
 		if addr == "" {
 			return fmt.Errorf("topic=%s route info not found", mq.Topic)
 		}
@@ -242,7 +354,7 @@ func (p *defaultProducer) sendAsync(ctx context.Context, msg *primitive.Message,
 		return errors.Errorf("the topic=%s route info not found", msg.Topic)
 	}
 
-	addr := p.options.Namesrv.FindBrokerAddrByName(mq.BrokerName)
+	addr := p.client.GetNameSrv().FindBrokerAddrByName(mq.BrokerName)
 	if addr == "" {
 		return errors.Errorf("topic=%s route info not found", mq.Topic)
 	}
@@ -271,7 +383,7 @@ func (p *defaultProducer) SendOneWay(ctx context.Context, msgs ...*primitive.Mes
 	if p.interceptor != nil {
 		primitive.WithMethod(ctx, primitive.SendOneway)
 		return p.interceptor(ctx, msg, nil, func(ctx context.Context, req, reply interface{}) error {
-			return p.SendOneWay(ctx, msg)
+			return p.sendOneWay(ctx, msg)
 		})
 	}
 
@@ -289,7 +401,7 @@ func (p *defaultProducer) sendOneWay(ctx context.Context, msg *primitive.Message
 			continue
 		}
 
-		addr := p.options.Namesrv.FindBrokerAddrByName(mq.BrokerName)
+		addr := p.client.GetNameSrv().FindBrokerAddrByName(mq.BrokerName)
 		if addr == "" {
 			return fmt.Errorf("topic=%s route info not found", mq.Topic)
 		}
@@ -329,7 +441,7 @@ func (p *defaultProducer) tryCompressMsg(msg *primitive.Message) bool {
 	if e != nil {
 		return false
 	}
-	msg.Body = compressedBody
+	msg.CompressedBody = compressedBody
 	msg.Compress = true
 	return true
 }
@@ -339,8 +451,14 @@ func (p *defaultProducer) buildSendRequest(mq *primitive.MessageQueue,
 	if !msg.Batch && msg.GetProperty(primitive.PropertyUniqueClientMessageIdKeyIndex) == "" {
 		msg.WithProperty(primitive.PropertyUniqueClientMessageIdKeyIndex, primitive.CreateUniqID())
 	}
-	sysFlag := 0
+
+	var (
+		sysFlag      = 0
+		transferBody = msg.Body
+	)
+
 	if p.tryCompressMsg(msg) {
+		transferBody = msg.CompressedBody
 		sysFlag = primitive.SetCompressedFlag(sysFlag)
 	}
 	v := msg.GetProperty(primitive.PropertyTransactionPrepared)
@@ -363,22 +481,26 @@ func (p *defaultProducer) buildSendRequest(mq *primitive.MessageQueue,
 		UnitMode:       p.options.UnitMode,
 		Batch:          msg.Batch,
 	}
+
+	msgType := msg.GetProperty(primitive.PropertyMsgType)
+	if msgType == internal.ReplyMessageFlag {
+		return remote.NewRemotingCommand(internal.ReqSendReplyMessage, req, msg.Body)
+	}
+
 	cmd := internal.ReqSendMessage
 	if msg.Batch {
 		cmd = internal.ReqSendBatchMessage
 		reqv2 := &internal.SendMessageRequestV2Header{SendMessageRequestHeader: req}
-		return remote.NewRemotingCommand(cmd, reqv2, msg.Body)
+		return remote.NewRemotingCommand(cmd, reqv2, transferBody)
 	}
 
-	return remote.NewRemotingCommand(cmd, req, msg.Body)
+	return remote.NewRemotingCommand(cmd, req, transferBody)
 }
 
-func (p *defaultProducer) selectMessageQueue(msg *primitive.Message) *primitive.MessageQueue {
-	topic := msg.Topic
-
+func (p *defaultProducer) tryToFindTopicPublishInfo(topic string) *internal.TopicPublishInfo {
 	v, exist := p.publishInfo.Load(topic)
 	if !exist {
-		data, changed, err := p.options.Namesrv.UpdateTopicRouteInfo(topic)
+		data, changed, err := p.client.GetNameSrv().UpdateTopicRouteInfo(topic)
 		if err != nil && primitive.IsRemotingErr(err) {
 			return nil
 		}
@@ -387,7 +509,7 @@ func (p *defaultProducer) selectMessageQueue(msg *primitive.Message) *primitive.
 	}
 
 	if !exist {
-		data, changed, _ := p.options.Namesrv.UpdateTopicRouteInfoWithDefault(topic, p.options.CreateTopicKey, p.options.DefaultTopicQueueNums)
+		data, changed, _ := p.client.GetNameSrv().UpdateTopicRouteInfoWithDefault(topic, p.options.CreateTopicKey, p.options.DefaultTopicQueueNums)
 		p.client.UpdatePublishInfo(topic, data, changed)
 		v, exist = p.publishInfo.Load(topic)
 	}
@@ -405,7 +527,12 @@ func (p *defaultProducer) selectMessageQueue(msg *primitive.Message) *primitive.
 		rlog.Error("can not find proper message queue", nil)
 		return nil
 	}
+	return result
+}
 
+func (p *defaultProducer) selectMessageQueue(msg *primitive.Message) *primitive.MessageQueue {
+	topic := msg.Topic
+	result := p.tryToFindTopicPublishInfo(topic)
 	return p.options.Selector.Select(msg, result.MqList)
 }
 
@@ -558,8 +685,8 @@ func (tp *transactionProducer) endTransaction(result primitive.SendResult, err e
 	} else {
 		msgID, _ = primitive.UnmarshalMsgID([]byte(result.MsgID))
 	}
-	
-	brokerAddr := tp.producer.options.Namesrv.FindBrokerAddrByName(result.MessageQueue.BrokerName)
+	// 估计没有反序列化回来
+	brokerAddr := tp.producer.client.GetNameSrv().FindBrokerAddrByName(result.MessageQueue.BrokerName)
 	requestHeader := &internal.EndTransactionRequestHeader{
 		TransactionId:        result.TransactionID,
 		CommitLogOffset:      msgID.Offset,
diff --git a/rlog/log.go b/rlog/log.go
index 037cfcf..b387ed7 100644
--- a/rlog/log.go
+++ b/rlog/log.go
@@ -19,12 +19,16 @@ package rlog
 
 import (
 	"os"
+	"path/filepath"
 	"strings"
 
+	"gopkg.in/natefinch/lumberjack.v2"
+
 	"github.com/sirupsen/logrus"
 )
 
 const (
+	LogKeyProducerGroup    = "producerGroup"
 	LogKeyConsumerGroup    = "consumerGroup"
 	LogKeyTopic            = "topic"
 	LogKeyMessageQueue     = "MessageQueue"
@@ -122,14 +126,49 @@ func (l *defaultLogger) Level(level string) {
 	}
 }
 
-func (l *defaultLogger) OutputPath(path string) (err error) {
-	var file *os.File
-	file, err = os.OpenFile(path, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0666)
-	if err != nil {
-		return
+type Config struct {
+	OutputPath    string
+	MaxFileSizeMB int
+	MaxBackups    int
+	MaxAges       int
+	Compress      bool
+	LocalTime     bool
+}
+
+func (c *Config) Logger() *lumberjack.Logger {
+	return &lumberjack.Logger{
+		Filename:   filepath.ToSlash(c.OutputPath),
+		MaxSize:    c.MaxFileSizeMB, // MB
+		MaxBackups: c.MaxBackups,
+		MaxAge:     c.MaxAges,  // days
+		Compress:   c.Compress, // disabled by default
+		LocalTime:  c.LocalTime,
+	}
+}
+
+const defaultLogPath = "/tmp/rocketmq-client.log"
+
+func defaultConfig() Config {
+	return Config{
+		OutputPath:    defaultLogPath,
+		MaxFileSizeMB: 10,
+		MaxBackups:    5,
+		MaxAges:       3,
+		Compress:      false,
+		LocalTime:     true,
 	}
+}
+
+func (l *defaultLogger) Config(conf Config) (err error) {
+	l.logger.Out = conf.Logger()
+	return
+}
+
+func (l *defaultLogger) OutputPath(path string) (err error) {
+	config := defaultConfig()
+	config.OutputPath = path
 
-	l.logger.Out = file
+	l.logger.Out = config.Logger()
 	return
 }
 
@@ -137,6 +176,7 @@ func (l *defaultLogger) OutputPath(path string) (err error) {
 func SetLogger(logger Logger) {
 	rLog = logger
 }
+
 func SetLogLevel(level string) {
 	if level == "" {
 		return


Test patch:
diff --git a/consumer/consumer_test.go b/consumer/consumer_test.go
index 8b99767..e441290 100644
--- a/consumer/consumer_test.go
+++ b/consumer/consumer_test.go
@@ -26,6 +26,7 @@ import (
 	. "github.com/smartystreets/goconvey/convey"
 	"github.com/stretchr/testify/assert"
 
+	"github.com/apache/rocketmq-client-go/v2/errors"
 	"github.com/apache/rocketmq-client-go/v2/internal"
 	"github.com/apache/rocketmq-client-go/v2/internal/remote"
 	"github.com/apache/rocketmq-client-go/v2/primitive"
@@ -67,7 +68,6 @@ func TestDoRebalance(t *testing.T) {
 		defer ctrl.Finish()
 		namesrvCli := internal.NewMockNamesrvs(ctrl)
 		namesrvCli.EXPECT().FindBrokerAddrByTopic(gomock.Any()).Return(broker)
-		dc.namesrv = namesrvCli
 
 		rmqCli := internal.NewMockRMQClient(ctrl)
 		rmqCli.EXPECT().InvokeSync(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
@@ -75,6 +75,8 @@ func TestDoRebalance(t *testing.T) {
 				Body: []byte("{\"consumerIdList\": [\"a1\", \"a2\", \"a3\"] }"),
 			}, nil)
 		rmqCli.EXPECT().ClientID().Return(clientID)
+		rmqCli.SetNameSrv(namesrvCli)
+
 		dc.client = rmqCli
 
 		var wg sync.WaitGroup
@@ -109,19 +111,26 @@ func TestComputePullFromWhere(t *testing.T) {
 		}
 
 		namesrvCli := internal.NewMockNamesrvs(ctrl)
-		dc.namesrv = namesrvCli
 
 		rmqCli := internal.NewMockRMQClient(ctrl)
 		dc.client = rmqCli
+		rmqCli.SetNameSrv(namesrvCli)
 
 		Convey("get effective offset", func() {
-			offsetStore.EXPECT().read(gomock.Any(), gomock.Any()).Return(int64(10))
-			res := dc.computePullFromWhere(mq)
+			offsetStore.EXPECT().readWithException(gomock.Any(), gomock.Any()).Return(int64(10), nil)
+			res, _ := dc.computePullFromWhereWithException(mq)
 			assert.Equal(t, int64(10), res)
 		})
 
+		Convey("get offset error", func() {
+			offsetStore.EXPECT().readWithException(gomock.Any(), gomock.Any()).Return(int64(-1), errors.ErrRequestTimeout)
+
+			_, err := dc.computePullFromWhereWithException(mq)
+			assert.Equal(t, err, errors.ErrRequestTimeout)
+		})
+
 		Convey("ConsumeFromLastOffset for normal topic", func() {
-			offsetStore.EXPECT().read(gomock.Any(), gomock.Any()).Return(int64(-1))
+			offsetStore.EXPECT().readWithException(gomock.Any(), gomock.Any()).Return(int64(-1), nil)
 			dc.option.FromWhere = ConsumeFromLastOffset
 
 			broker := "a"
@@ -134,20 +143,20 @@ func TestComputePullFromWhere(t *testing.T) {
 					},
 				}, nil)
 
-			res := dc.computePullFromWhere(mq)
+			res, _ := dc.computePullFromWhereWithException(mq)
 			assert.Equal(t, int64(20), res)
 		})
 
 		Convey("ConsumeFromFirstOffset for normal topic", func() {
-			offsetStore.EXPECT().read(gomock.Any(), gomock.Any()).Return(int64(-1))
+			offsetStore.EXPECT().readWithException(gomock.Any(), gomock.Any()).Return(int64(-1), nil)
 			dc.option.FromWhere = ConsumeFromFirstOffset
 
-			res := dc.computePullFromWhere(mq)
+			res, _ := dc.computePullFromWhereWithException(mq)
 			assert.Equal(t, int64(0), res)
 		})
 
 		Convey("ConsumeFromTimestamp for normal topic", func() {
-			offsetStore.EXPECT().read(gomock.Any(), gomock.Any()).Return(int64(-1))
+			offsetStore.EXPECT().readWithException(gomock.Any(), gomock.Any()).Return(int64(-1), nil)
 			dc.option.FromWhere = ConsumeFromTimestamp
 
 			dc.option.ConsumeTimestamp = "20060102150405"
@@ -162,7 +171,7 @@ func TestComputePullFromWhere(t *testing.T) {
 					},
 				}, nil)
 
-			res := dc.computePullFromWhere(mq)
+			res, _ := dc.computePullFromWhereWithException(mq)
 			assert.Equal(t, int64(30), res)
 		})
 
diff --git a/consumer/offset_store_test.go b/consumer/offset_store_test.go
index cfa0eaa..d833f36 100644
--- a/consumer/offset_store_test.go
+++ b/consumer/offset_store_test.go
@@ -98,7 +98,7 @@ func TestLocalFileOffsetStore(t *testing.T) {
 				}
 				for _, value := range cases {
 					localStore.update(value.queue, value.setOffset, false)
-					offset := localStore.read(value.queue, _ReadFromMemory)
+					offset, _ := localStore.readWithException(value.queue, _ReadFromMemory)
 					So(offset, ShouldEqual, value.expectedOffset)
 				}
 			})
@@ -119,7 +119,7 @@ func TestLocalFileOffsetStore(t *testing.T) {
 				}
 				for _, value := range cases {
 					localStore.update(value.queue, value.setOffset, true)
-					offset := localStore.read(value.queue, _ReadFromMemory)
+					offset, _ := localStore.readWithException(value.queue, _ReadFromMemory)
 					So(offset, ShouldEqual, value.expectedOffset)
 				}
 			})
@@ -127,16 +127,16 @@ func TestLocalFileOffsetStore(t *testing.T) {
 
 		Convey("test persist", func() {
 			localStore.update(mq, 1, false)
-			offset := localStore.read(mq, _ReadFromMemory)
+			offset, _ := localStore.readWithException(mq, _ReadFromMemory)
 			So(offset, ShouldEqual, 1)
 
 			queues := []*primitive.MessageQueue{mq}
 			localStore.persist(queues)
-			offset = localStore.read(mq, _ReadFromStore)
+			offset, _ = localStore.readWithException(mq, _ReadFromStore)
 			So(offset, ShouldEqual, 1)
 
 			localStore.(*localFileOffsetStore).OffsetTable.Delete(MessageQueueKey(*mq))
-			offset = localStore.read(mq, _ReadMemoryThenStore)
+			offset, _ = localStore.readWithException(mq, _ReadMemoryThenStore)
 			So(offset, ShouldEqual, 1)
 		})
 	})
@@ -178,7 +178,7 @@ func TestRemoteBrokerOffsetStore(t *testing.T) {
 				}
 				for _, value := range cases {
 					remoteStore.update(value.queue, value.setOffset, false)
-					offset := remoteStore.read(value.queue, _ReadFromMemory)
+					offset, _ := remoteStore.readWithException(value.queue, _ReadFromMemory)
 					So(offset, ShouldEqual, value.expectedOffset)
 				}
 			})
@@ -199,7 +199,7 @@ func TestRemoteBrokerOffsetStore(t *testing.T) {
 				}
 				for _, value := range cases {
 					remoteStore.update(value.queue, value.setOffset, true)
-					offset := remoteStore.read(value.queue, _ReadFromMemory)
+					offset, _ := remoteStore.readWithException(value.queue, _ReadFromMemory)
 					So(offset, ShouldEqual, value.expectedOffset)
 				}
 			})
@@ -219,24 +219,24 @@ func TestRemoteBrokerOffsetStore(t *testing.T) {
 			rmqClient.EXPECT().InvokeSync(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(ret, nil).MaxTimes(2)
 
 			remoteStore.persist(queues)
-			offset := remoteStore.read(mq, _ReadFromStore)
+			offset, _ := remoteStore.readWithException(mq, _ReadFromStore)
 			So(offset, ShouldEqual, 1)
 
 			remoteStore.remove(mq)
-			offset = remoteStore.read(mq, _ReadFromMemory)
+			offset, _ = remoteStore.readWithException(mq, _ReadFromMemory)
 			So(offset, ShouldEqual, -1)
-			offset = remoteStore.read(mq, _ReadMemoryThenStore)
+			offset, _ = remoteStore.readWithException(mq, _ReadMemoryThenStore)
 			So(offset, ShouldEqual, 1)
 
 		})
 
 		Convey("test remove", func() {
 			remoteStore.update(mq, 1, false)
-			offset := remoteStore.read(mq, _ReadFromMemory)
+			offset, _ := remoteStore.readWithException(mq, _ReadFromMemory)
 			So(offset, ShouldEqual, 1)
 
 			remoteStore.remove(mq)
-			offset = remoteStore.read(mq, _ReadFromMemory)
+			offset, _ = remoteStore.readWithException(mq, _ReadFromMemory)
 			So(offset, ShouldEqual, -1)
 		})
 	})
diff --git a/consumer/strategy_test.go b/consumer/strategy_test.go
index d521b4b..c50f403 100644
--- a/consumer/strategy_test.go
+++ b/consumer/strategy_test.go
@@ -477,8 +477,8 @@ func TestAllocateByConsistentHash(t *testing.T) {
 			for _, value := range cases {
 				result := strategy("testGroup", value.currentCid, value.mqAll, value.cidAll)
 				rlog.Info("Result Of AllocateByMachineRoom", map[string]interface{}{
-					"currentCid": value.currentCid,
-					"cidAll": value.cidAll,
+					"currentCid":     value.currentCid,
+					"cidAll":         value.cidAll,
 					"allocateResult": result,
 				})
 			}
diff --git a/internal/namesrv_test.go b/internal/namesrv_test.go
index b047a07..27e15e3 100644
--- a/internal/namesrv_test.go
+++ b/internal/namesrv_test.go
@@ -36,7 +36,7 @@ import (
 // TestSelector test roundrobin selector in namesrv
 func TestSelector(t *testing.T) {
 	srvs := []string{"127.0.0.1:9876", "127.0.0.1:9879", "12.24.123.243:10911", "12.24.123.243:10915"}
-	namesrv, err := NewNamesrv(primitive.NewPassthroughResolver(srvs))
+	namesrv, err := NewNamesrv(primitive.NewPassthroughResolver(srvs), nil)
 	assert.Nil(t, err)
 
 	assert.Equal(t, srvs[0], namesrv.getNameServerAddress())
diff --git a/internal/remote/codec_test.go b/internal/remote/codec_test.go
index 0717451..60f7869 100644
--- a/internal/remote/codec_test.go
+++ b/internal/remote/codec_test.go
@@ -364,7 +364,7 @@ func TestCommandJsonIter(t *testing.T) {
 	err = json.Unmarshal(cmdData, &cmdResp)
 	assert.Nil(t, err)
 	rlog.Info("Json Decode Success", map[string]interface{}{
-		"cmd": cmdResp,
+		"cmd":      cmdResp,
 		"language": cmdResp.Language,
 	})
 
@@ -372,7 +372,7 @@ func TestCommandJsonIter(t *testing.T) {
 	err = json.Unmarshal(data, &cmdResp2)
 	assert.Nil(t, err)
 	rlog.Info("Json Decode Success", map[string]interface{}{
-		"cmd": cmdResp2,
+		"cmd":      cmdResp2,
 		"language": cmdResp2.Language,
 	})
 }
diff --git a/internal/remote/remote_client_test.go b/internal/remote/remote_client_test.go
index fa33f4f..b7171c4 100644
--- a/internal/remote/remote_client_test.go
+++ b/internal/remote/remote_client_test.go
@@ -116,7 +116,7 @@ func TestCreateScanner(t *testing.T) {
 	if err != nil {
 		t.Fatalf("failed to encode RemotingCommand. %s", err)
 	}
-	client := NewRemotingClient()
+	client := NewRemotingClient(nil)
 	reader := bytes.NewReader(content)
 	scanner := client.createScanner(reader)
 	for scanner.Scan() {
@@ -151,7 +151,7 @@ func TestInvokeSync(t *testing.T) {
 	serverSendRemotingCommand.Flag = ResponseType
 	var wg sync.WaitGroup
 	wg.Add(1)
-	client := NewRemotingClient()
+	client := NewRemotingClient(nil)
 
 	var clientSend sync.WaitGroup // blocking client send message until the server listen success.
 	clientSend.Add(1)
@@ -216,7 +216,7 @@ func TestInvokeAsync(t *testing.T) {
 	var wg sync.WaitGroup
 	cnt := 50
 	wg.Add(cnt)
-	client := NewRemotingClient()
+	client := NewRemotingClient(nil)
 	for i := 0; i < cnt; i++ {
 		go func(index int) {
 			time.Sleep(time.Duration(rand.Intn(100)) * time.Millisecond)
@@ -282,7 +282,7 @@ func TestInvokeAsyncTimeout(t *testing.T) {
 
 	var wg sync.WaitGroup
 	wg.Add(1)
-	client := NewRemotingClient()
+	client := NewRemotingClient(nil)
 
 	var clientSend sync.WaitGroup // blocking client send message until the server listen success.
 	clientSend.Add(1)
@@ -329,7 +329,7 @@ func TestInvokeOneWay(t *testing.T) {
 
 	var wg sync.WaitGroup
 	wg.Add(1)
-	client := NewRemotingClient()
+	client := NewRemotingClient(nil)
 
 	var clientSend sync.WaitGroup // blocking client send message until the server listen success.
 	clientSend.Add(1)
diff --git a/internal/route_test.go b/internal/route_test.go
index a1ebec4..020d8d2 100644
--- a/internal/route_test.go
+++ b/internal/route_test.go
@@ -42,7 +42,7 @@ func TestQueryTopicRouteInfoFromServer(t *testing.T) {
 		addr, err := primitive.NewNamesrvAddr("1.1.1.1:8880", "1.1.1.2:8880", "1.1.1.3:8880")
 		assert.Nil(t, err)
 
-		namesrv, err := NewNamesrv(primitive.NewPassthroughResolver(addr))
+		namesrv, err := NewNamesrv(primitive.NewPassthroughResolver(addr), nil)
 		assert.Nil(t, err)
 		namesrv.nameSrvClient = remotingCli
 
diff --git a/producer/producer_test.go b/producer/producer_test.go
index a7c15c1..e1d72dd 100644
--- a/producer/producer_test.go
+++ b/producer/producer_test.go
@@ -47,7 +47,7 @@ func TestShutdown(t *testing.T) {
 	client := internal.NewMockRMQClient(ctrl)
 	p.client = client
 
-	client.EXPECT().RegisterProducer(gomock.Any(), gomock.Any()).Return()
+	client.EXPECT().RegisterProducer(gomock.Any(), gomock.Any()).Return(nil)
 	client.EXPECT().Start().Return()
 	err := p.Start()
 	assert.Nil(t, err)
@@ -117,10 +117,13 @@ func TestSync(t *testing.T) {
 
 	ctrl := gomock.NewController(t)
 	defer ctrl.Finish()
+	namesrvCli := internal.NewMockNamesrvs(ctrl)
 	client := internal.NewMockRMQClient(ctrl)
 	p.client = client
+	client.SetNameSrv(namesrvCli)
+	namesrvCli.EXPECT().FindBrokerAddrByName(gomock.Any()).Return("a")
 
-	client.EXPECT().RegisterProducer(gomock.Any(), gomock.Any()).Return()
+	client.EXPECT().RegisterProducer(gomock.Any(), gomock.Any()).Return(nil)
 	client.EXPECT().Start().Return()
 	err := p.Start()
 	assert.Nil(t, err)
@@ -168,10 +171,13 @@ func TestASync(t *testing.T) {
 
 	ctrl := gomock.NewController(t)
 	defer ctrl.Finish()
+	namesrvCli := internal.NewMockNamesrvs(ctrl)
 	client := internal.NewMockRMQClient(ctrl)
 	p.client = client
+	client.SetNameSrv(namesrvCli)
+	namesrvCli.EXPECT().FindBrokerAddrByName(gomock.Any()).Return("a")
 
-	client.EXPECT().RegisterProducer(gomock.Any(), gomock.Any()).Return()
+	client.EXPECT().RegisterProducer(gomock.Any(), gomock.Any()).Return(nil)
 	client.EXPECT().Start().Return()
 	err := p.Start()
 	assert.Nil(t, err)
@@ -230,10 +236,13 @@ func TestOneway(t *testing.T) {
 
 	ctrl := gomock.NewController(t)
 	defer ctrl.Finish()
+	namesrvCli := internal.NewMockNamesrvs(ctrl)
 	client := internal.NewMockRMQClient(ctrl)
 	p.client = client
+	client.SetNameSrv(namesrvCli)
+	namesrvCli.EXPECT().FindBrokerAddrByName(gomock.Any()).Return("a")
 
-	client.EXPECT().RegisterProducer(gomock.Any(), gomock.Any()).Return()
+	client.EXPECT().RegisterProducer(gomock.Any(), gomock.Any()).Return(nil)
 	client.EXPECT().Start().Return()
 	err := p.Start()
 	assert.Nil(t, err)
@@ -268,10 +277,13 @@ func TestSyncWithNamespace(t *testing.T) {
 
 	ctrl := gomock.NewController(t)
 	defer ctrl.Finish()
+	namesrvCli := internal.NewMockNamesrvs(ctrl)
 	client := internal.NewMockRMQClient(ctrl)
 	p.client = client
+	client.SetNameSrv(namesrvCli)
+	namesrvCli.EXPECT().FindBrokerAddrByName(gomock.Any()).Return("a")
 
-	client.EXPECT().RegisterProducer(gomock.Any(), gomock.Any()).Return()
+	client.EXPECT().RegisterProducer(gomock.Any(), gomock.Any()).Return(nil)
 	client.EXPECT().Start().Return()
 	err := p.Start()
 	assert.Nil(t, err)
@@ -323,7 +335,7 @@ func TestBatchSendDifferentTopics(t *testing.T) {
 	client := internal.NewMockRMQClient(ctrl)
 	p.client = client
 
-	client.EXPECT().RegisterProducer(gomock.Any(), gomock.Any()).Return()
+	client.EXPECT().RegisterProducer(gomock.Any(), gomock.Any()).Return(nil)
 	client.EXPECT().Start().Return()
 	err := p.Start()
 	assert.Nil(t, err)