Repository: apache/rocketmq-client-go (go)

Issue:
It will stop to retry when use SyncSend to send in some case
**BUG REPORT**  

Currently in SyncSend, retries are only worked when there is a network error, but not for common errors such as Broker busy
```golang
res, _err := p.client.InvokeSync(ctx, addr, p.buildSendRequest(mq, msg), 3*time.Second)
if _err != nil {
	err = _err
	continue
}

return p.client.ProcessSendResponse(mq.BrokerName, res, resp, msg)
```
In [java SDK](https://github.com/apache/rocketmq/blob/dcf7ccdd9ba92c22a44cde4c1df36c4d1ef6a172/client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java#LL615C17-L615C17), It will retry for `broker_busy`、`servicce not available` and so on. And It will select a different broker to retry. 

**This is very helpful in a multi-Broker cluster, so I think it is necessary to fix this problem.**


Pull request description:
[ISSUE #1013] Fix: select one message queue of different broker when retry to send
## What is the purpose of the change

fix #1013 

## Brief changelog

XX

## Verifying this change

XXXX

Follow this checklist to help us incorporate your contribution quickly and easily. Notice, `it would be helpful if you could finish the following 5 checklist(the last one is not necessary)before request the community to review your PR`.

- [x] Make sure there is a [Github issue](https://github.com/apache/rocketmq/issues) filed for the change (usually before you start working on it). Trivial changes like typos do not require a Github issue. Your pull request should address just this issue, without pulling in other changes - one PR resolves one issue. 
- [x] Format the pull request title like `[ISSUE #123] Fix UnknownException when host config not exist`. Each commit in the pull request should have a meaningful subject line and body.
- [x] Write a pull request description that is detailed enough to understand what the pull request does, how, and why.
- [x] Write necessary unit-test(over 80% coverage) to verify your logic correction, more mock a little better when a cross-module dependency exists.
- [ ] If this contribution is large, please file an [Apache Individual Contributor License Agreement](http://www.apache.org/licenses/#clas).


Patch:
diff --git a/consumer/mock_offset_store.go b/consumer/mock_offset_store.go
index 145ec3d..1d7c3bd 100644
--- a/consumer/mock_offset_store.go
+++ b/consumer/mock_offset_store.go
@@ -115,3 +115,17 @@ func (mr *MockOffsetStoreMockRecorder) update(mq, offset, increaseOnly interface
 	mr.mock.ctrl.T.Helper()
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "update", reflect.TypeOf((*MockOffsetStore)(nil).update), mq, offset, increaseOnly)
 }
+
+// getMQOffsetMap mocks base method
+func (m *MockOffsetStore) getMQOffsetMap(topic string) map[primitive.MessageQueue]int64  {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "getMQOffsetMap", topic)
+	ret0, _ := ret[0].(map[primitive.MessageQueue]int64)
+	return ret0
+}
+
+// getMQOffsetMap indicates an expected call of getMQOffsetMap
+func (mr *MockOffsetStoreMockRecorder) getMQOffsetMap(topic string) *gomock.Call{
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "getMQOffsetMap", reflect.TypeOf((*MockOffsetStore)(nil).getMQOffsetMap), topic)
+}
diff --git a/consumer/offset_store.go b/consumer/offset_store.go
index 0fa748c..b0f9da9 100644
--- a/consumer/offset_store.go
+++ b/consumer/offset_store.go
@@ -59,6 +59,7 @@ type OffsetStore interface {
 	read(mq *primitive.MessageQueue, t readType) int64
 	readWithException(mq *primitive.MessageQueue, t readType) (int64, error)
 	update(mq *primitive.MessageQueue, offset int64, increaseOnly bool)
+	getMQOffsetMap(topic string) map[primitive.MessageQueue]int64
 }
 
 type OffsetSerializeWrapper struct {
@@ -228,6 +229,18 @@ func (local *localFileOffsetStore) remove(mq *primitive.MessageQueue) {
 	// nothing to do
 }
 
+func (local *localFileOffsetStore) getMQOffsetMap(topic string) map[primitive.MessageQueue]int64 {
+	copyOffsetTable := make(map[primitive.MessageQueue]int64)
+	local.OffsetTable.Range(func(key, value interface{}) bool {
+		if key.(MessageQueueKey).Topic != topic {
+			return true
+		}
+		copyOffsetTable[primitive.MessageQueue(key.(MessageQueueKey))] = value.(int64)
+		return true
+	})
+	return copyOffsetTable
+}
+
 type remoteBrokerOffsetStore struct {
 	group       string
 	OffsetTable map[primitive.MessageQueue]int64 `json:"OffsetTable"`
@@ -353,6 +366,17 @@ func (r *remoteBrokerOffsetStore) update(mq *primitive.MessageQueue, offset int6
 	}
 }
 
+func (r *remoteBrokerOffsetStore) getMQOffsetMap(topic string) map[primitive.MessageQueue]int64 {
+	copyOffsetTable := make(map[primitive.MessageQueue]int64)
+	for mq, offset := range r.OffsetTable {
+		if mq.Topic != topic {
+			continue
+		}
+		copyOffsetTable[mq] = offset
+	}
+	return copyOffsetTable
+}
+
 func (r *remoteBrokerOffsetStore) fetchConsumeOffsetFromBroker(group string, mq *primitive.MessageQueue) (int64, error) {
 	broker := r.namesrv.FindBrokerAddrByName(mq.BrokerName)
 	if broker == "" {
diff --git a/consumer/pull_consumer.go b/consumer/pull_consumer.go
index ef165b0..6730f07 100644
--- a/consumer/pull_consumer.go
+++ b/consumer/pull_consumer.go
@@ -853,6 +853,15 @@ func (pc *defaultPullConsumer) consumeMessageConcurrently(pq *processQueue, mq *
 	}
 }
 
+func (pc *defaultPullConsumer) GetConsumerStatus(topic string) *internal.ConsumerStatus {
+	consumerStatus := internal.NewConsumerStatus()
+	mqOffsetMap := pc.storage.getMQOffsetMap(topic)
+	if mqOffsetMap != nil {
+		consumerStatus.MQOffsetMap = mqOffsetMap
+	}
+	return consumerStatus
+}
+
 func (pc *defaultPullConsumer) validate() error {
 	if err := internal.ValidateGroup(pc.consumerGroup); err != nil {
 		return err
diff --git a/consumer/push_consumer.go b/consumer/push_consumer.go
index 531a5b6..50eee98 100644
--- a/consumer/push_consumer.go
+++ b/consumer/push_consumer.go
@@ -430,6 +430,15 @@ func (pc *pushConsumer) ConsumeMessageDirectly(msg *primitive.MessageExt, broker
 	return res
 }
 
+func (pc *pushConsumer) GetConsumerStatus(topic string) *internal.ConsumerStatus {
+	consumerStatus := internal.NewConsumerStatus()
+	mqOffsetMap := pc.storage.getMQOffsetMap(topic)
+	if mqOffsetMap != nil {
+		consumerStatus.MQOffsetMap = mqOffsetMap
+	}
+	return consumerStatus
+}
+
 func (pc *pushConsumer) GetConsumerRunningInfo(stack bool) *internal.ConsumerRunningInfo {
 	info := internal.NewConsumerRunningInfo()
 
@@ -540,7 +549,9 @@ func (pc *pushConsumer) validate() error {
 	}
 
 	if len(pc.subscribedTopic) == 0 {
-		return errors.New("number of subscribed topics is 0.")
+		rlog.Warning("not subscribe any topic yet", map[string]interface{}{
+			rlog.LogKeyConsumerGroup: pc.consumerGroup,
+		})
 	}
 
 	if pc.option.ConsumeConcurrentlyMaxSpan < 1 || pc.option.ConsumeConcurrentlyMaxSpan > 65535 {
@@ -603,11 +614,11 @@ func (pc *pushConsumer) validate() error {
 		}
 	}
 
-	if pc.option.ConsumeGoroutineNums < 1 || pc.option.ConsumeGoroutineNums > 1000 {
+	if pc.option.ConsumeGoroutineNums < 1 || pc.option.ConsumeGoroutineNums > 100000 {
 		if pc.option.ConsumeGoroutineNums == 0 {
 			pc.option.ConsumeGoroutineNums = 20
 		} else {
-			return errors.New("option.ConsumeGoroutineNums out of range [1, 1000]")
+			return errors.New("option.ConsumeGoroutineNums out of range [1, 100000]")
 		}
 	}
 	return nil
diff --git a/internal/client.go b/internal/client.go
index f7b2aec..1321da9 100644
--- a/internal/client.go
+++ b/internal/client.go
@@ -19,7 +19,6 @@ package internal
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"net"
 	"os"
@@ -93,6 +92,7 @@ type InnerConsumer interface {
 	GetModel() string
 	GetWhere() string
 	ResetOffset(topic string, table map[primitive.MessageQueue]int64)
+	GetConsumerStatus(topic string) *ConsumerStatus
 }
 
 func DefaultClientOptions() ClientOptions {
@@ -378,6 +378,32 @@ func GetOrNewRocketMQClient(option ClientOptions, callbackCh chan interface{}) R
 			res.Code = ResSuccess
 			return res
 		})
+
+		client.remoteClient.RegisterRequestFunc(ReqGetConsumerStatsFromClient, func(req *remote.RemotingCommand, addr net.Addr) *remote.RemotingCommand {
+			rlog.Info("receive get consumer status from client request...", map[string]interface{}{
+				rlog.LogKeyBroker:        addr.String(),
+				rlog.LogKeyTopic:         req.ExtFields["topic"],
+				rlog.LogKeyConsumerGroup: req.ExtFields["group"],
+			})
+
+			header := new(GetConsumerStatusRequestHeader)
+			header.Decode(req.ExtFields)
+			res := remote.NewRemotingCommand(ResError, nil, nil)
+
+			consumerStatus := client.getConsumerStatus(header.topic, header.group)
+			if consumerStatus != nil {
+				res.Code = ResSuccess
+				data, err := consumerStatus.Encode()
+				if err != nil {
+					res.Remark = fmt.Sprintf("Failed to encode consumer status: %s", err.Error())
+				} else {
+					res.Body = data
+				}
+			} else {
+				res.Remark = "there is unexpected error when get consumer status, please check log"
+			}
+			return res
+		})
 	}
 	return client
 }
@@ -385,10 +411,8 @@ func GetOrNewRocketMQClient(option ClientOptions, callbackCh chan interface{}) R
 func (c *rmqClient) Start() {
 	//ctx, cancel := context.WithCancel(context.Background())
 	//c.cancel = cancel
+	atomic.AddInt32(&c.instanceCount, 1)
 	c.once.Do(func() {
-
-		atomic.AddInt32(&c.instanceCount, 1)
-
 		if !c.option.Credentials.IsEmpty() {
 			c.remoteClient.RegisterInterceptor(remote.ACLInterceptor(c.option.Credentials))
 		}
@@ -662,19 +686,17 @@ func (c *rmqClient) SendHeartbeatToAllBrokerWithLock() {
 }
 
 func (c *rmqClient) UpdateTopicRouteInfo() {
+	allTopics := make(map[string]bool, 0)
 	publishTopicSet := make(map[string]bool, 0)
 	c.producerMap.Range(func(key, value interface{}) bool {
 		producer := value.(InnerProducer)
 		list := producer.PublishTopicList()
 		for idx := range list {
 			publishTopicSet[list[idx]] = true
+			allTopics[list[idx]] = true
 		}
 		return true
 	})
-	for topic := range publishTopicSet {
-		data, changed, _ := c.GetNameSrv().UpdateTopicRouteInfo(topic)
-		c.UpdatePublishInfo(topic, data, changed)
-	}
 
 	subscribedTopicSet := make(map[string]bool, 0)
 	c.consumerMap.Range(func(key, value interface{}) bool {
@@ -682,13 +704,22 @@ func (c *rmqClient) UpdateTopicRouteInfo() {
 		list := consumer.SubscriptionDataList()
 		for idx := range list {
 			subscribedTopicSet[list[idx].Topic] = true
+			allTopics[list[idx].Topic] = true
 		}
 		return true
 	})
 
-	for topic := range subscribedTopicSet {
+	for topic := range allTopics {
 		data, changed, _ := c.GetNameSrv().UpdateTopicRouteInfo(topic)
-		c.updateSubscribeInfo(topic, data, changed)
+
+		if publishTopicSet[topic] {
+			c.UpdatePublishInfo(topic, data, changed)
+		}
+
+		if subscribedTopicSet[topic] {
+			c.updateSubscribeInfo(topic, data, changed)
+		}
+
 	}
 }
 
@@ -704,7 +735,7 @@ func (c *rmqClient) ProcessSendResponse(brokerName string, cmd *remote.RemotingC
 	case ResSuccess:
 		status = primitive.SendOK
 	default:
-		return errors.New(fmt.Sprintf("CODE: %d, DESC: %s", cmd.Code, cmd.Remark))
+		return fmt.Errorf("CODE: %d, DESC: %s", cmd.Code, cmd.Remark)
 	}
 
 	msgIDs := make([]string, 0)
@@ -908,6 +939,15 @@ func (c *rmqClient) resetOffset(topic string, group string, offsetTable map[prim
 	consumer.(InnerConsumer).ResetOffset(topic, offsetTable)
 }
 
+func (c *rmqClient) getConsumerStatus(topic string, group string) *ConsumerStatus {
+	consumer, exist := c.consumerMap.Load(group)
+	if !exist {
+		rlog.Warning("group "+group+" do not exists", nil)
+		return nil
+	}
+	return consumer.(InnerConsumer).GetConsumerStatus(topic)
+}
+
 func (c *rmqClient) getConsumerRunningInfo(group string, stack bool) *ConsumerRunningInfo {
 	consumer, exist := c.consumerMap.Load(group)
 	if !exist {
diff --git a/internal/model.go b/internal/model.go
index 0fd7955..36e06a1 100644
--- a/internal/model.go
+++ b/internal/model.go
@@ -73,15 +73,15 @@ func (sd *SubscriptionData) Clone() *SubscriptionData {
 
 	if sd.Tags.Items() != nil {
 		cloned.Tags = utils.NewSet()
-		for _, value := range sd.Tags.Items() {
-			cloned.Tags.Add(value)
+		for key, value := range sd.Tags.Items() {
+			cloned.Tags.AddKV(key, value.UniqueID())
 		}
 	}
 
 	if sd.Codes.Items() != nil {
 		cloned.Codes = utils.NewSet()
-		for _, value := range sd.Codes.Items() {
-			cloned.Codes.Add(value)
+		for key, value := range sd.Codes.Items() {
+			cloned.Codes.AddKV(key, value.UniqueID())
 		}
 	}
 
@@ -295,6 +295,54 @@ func NewConsumerRunningInfo() *ConsumerRunningInfo {
 	}
 }
 
+type ConsumerStatus struct {
+	MQOffsetMap map[primitive.MessageQueue]int64
+}
+
+func (status ConsumerStatus) Encode() ([]byte, error) {
+	mapJson := ""
+	keys := make([]primitive.MessageQueue, 0)
+
+	for k := range status.MQOffsetMap {
+		keys = append(keys, k)
+	}
+
+	sort.Slice(keys, func(i, j int) bool {
+		q1 := keys[i]
+		q2 := keys[j]
+		com := strings.Compare(q1.Topic, q2.Topic)
+		if com != 0 {
+			return com < 0
+		}
+
+		com = strings.Compare(q1.BrokerName, q2.BrokerName)
+		if com != 0 {
+			return com < 0
+		}
+
+		return q1.QueueId < q2.QueueId
+	})
+
+	for idx := range keys {
+		dataK, err := json.Marshal(keys[idx])
+		if err != nil {
+			return nil, err
+		}
+		dataV, err := json.Marshal(status.MQOffsetMap[keys[idx]])
+		mapJson = fmt.Sprintf("%s,%s:%s", mapJson, string(dataK), string(dataV))
+	}
+	mapJson = strings.TrimLeft(mapJson, ",")
+	jsonData := fmt.Sprintf("{\"%s\":%s}",
+		"messageQueueTable", fmt.Sprintf("{%s}", mapJson))
+	return []byte(jsonData), nil
+}
+
+func NewConsumerStatus() *ConsumerStatus {
+	return &ConsumerStatus{
+		MQOffsetMap: make(map[primitive.MessageQueue]int64),
+	}
+}
+
 type ConsumeMessageDirectlyResult struct {
 	Order          bool          `json:"order"`
 	AutoCommit     bool          `json:"autoCommit"`
diff --git a/internal/request.go b/internal/request.go
index 9a590b8..7e86b50 100644
--- a/internal/request.go
+++ b/internal/request.go
@@ -50,6 +50,7 @@ const (
 	ReqDeleteTopicInBroker           = int16(215)
 	ReqDeleteTopicInNameSrv          = int16(216)
 	ReqResetConsumerOffset           = int16(220)
+	ReqGetConsumerStatsFromClient    = int16(221)
 	ReqGetConsumerRunningInfo        = int16(307)
 	ReqConsumeMessageDirectly        = int16(309)
 	ReqSendReplyMessage              = int16(324)
@@ -490,6 +491,38 @@ func (request *ConsumeMessageDirectlyHeader) Decode(properties map[string]string
 	}
 }
 
+type GetConsumerStatusRequestHeader struct {
+	topic      string
+	group      string
+	clientAddr string
+}
+
+func (request *GetConsumerStatusRequestHeader) Encode() map[string]string {
+	return map[string]string{
+		"topic":      request.topic,
+		"group":      request.group,
+		"clientAddr": request.clientAddr,
+	}
+}
+
+func (request *GetConsumerStatusRequestHeader) Decode(properties map[string]string) {
+	if len(properties) == 0 {
+		return
+	}
+
+	if v, existed := properties["topic"]; existed {
+		request.topic = v
+	}
+
+	if v, existed := properties["group"]; existed {
+		request.group = v
+	}
+
+	if v, existed := properties["clientAddr"]; existed {
+		request.clientAddr = v
+	}
+}
+
 type ReplyMessageRequestHeader struct {
 	producerGroup         string
 	topic                 string
diff --git a/internal/response.go b/internal/response.go
index d1c7e18..8a815ba 100644
--- a/internal/response.go
+++ b/internal/response.go
@@ -23,6 +23,8 @@ const (
 	ResFlushDiskTimeout     = int16(10)
 	ResSlaveNotAvailable    = int16(11)
 	ResFlushSlaveTimeout    = int16(12)
+	ResServiceNotAvailable  = int16(14)
+	ResNoPermission         = int16(16)
 	ResTopicNotExist        = int16(17)
 	ResPullNotFound         = int16(19)
 	ResPullRetryImmediately = int16(20)
diff --git a/primitive/message.go b/primitive/message.go
index 32adc2b..f206cb5 100644
--- a/primitive/message.go
+++ b/primitive/message.go
@@ -182,13 +182,7 @@ func (m *Message) WithTag(tags string) *Message {
 }
 
 func (m *Message) WithKeys(keys []string) *Message {
-	var sb strings.Builder
-	for _, k := range keys {
-		sb.WriteString(k)
-		sb.WriteString(PropertyKeySeparator)
-	}
-
-	m.WithProperty(PropertyKeys, sb.String())
+	m.WithProperty(PropertyKeys, strings.Join(keys, PropertyKeySeparator))
 	return m
 }
 
diff --git a/primitive/nsresolver.go b/primitive/nsresolver.go
index 970242f..67ca7a7 100644
--- a/primitive/nsresolver.go
+++ b/primitive/nsresolver.go
@@ -118,7 +118,11 @@ func (h *HttpResolver) DomainWithUnit(unitName string) {
 	if strings.Contains(h.domain, "?nofix=1") {
 		return
 	}
-	h.domain = fmt.Sprintf("%s-%s?nofix=1", h.domain, unitName)
+	if strings.Contains(h.domain, "?") {
+		h.domain = strings.Replace(h.domain, "?", fmt.Sprintf("-%s?nofix=1&", unitName), 1)
+	} else {
+		h.domain = fmt.Sprintf("%s-%s?nofix=1", h.domain, unitName)
+	}
 }
 
 func (h *HttpResolver) Resolve() []string {
diff --git a/producer/producer.go b/producer/producer.go
index ebd5e33..d8ca54b 100644
--- a/producer/producer.go
+++ b/producer/producer.go
@@ -155,6 +155,21 @@ func MarshalMessageBatch(msgs ...*primitive.Message) []byte {
 	return buffer.Bytes()
 }
 
+func needRetryCode(code int16) bool {
+	switch code {
+	case internal.ResTopicNotExist:
+		return true
+	case internal.ResServiceNotAvailable:
+		return true
+	case internal.ResError:
+		return true
+	case internal.ResNoPermission:
+		return true
+	default:
+		return false
+	}
+}
+
 func (p *defaultProducer) prepareSendRequest(msg *primitive.Message, ttl time.Duration) (string, error) {
 	correlationId := uuid.New().String()
 	requestClientId := p.client.ClientID()
@@ -301,6 +316,7 @@ func (p *defaultProducer) sendSync(ctx context.Context, msg *primitive.Message,
 
 	var (
 		err error
+		mq  *primitive.MessageQueue
 	)
 
 	var (
@@ -308,12 +324,23 @@ func (p *defaultProducer) sendSync(ctx context.Context, msg *primitive.Message,
 		ok          bool
 	)
 	for retryCount := 0; retryCount < retryTime; retryCount++ {
-		mq := p.selectMessageQueue(msg)
+		var lastBrokerName string
+		if mq != nil {
+			lastBrokerName = mq.BrokerName
+		}
+		mq := p.selectMessageQueue(msg, lastBrokerName)
 		if mq == nil {
 			err = fmt.Errorf("the topic=%s route info not found", msg.Topic)
 			continue
 		}
 
+		if lastBrokerName != "" {
+			rlog.Warning("start retrying to send, ", map[string]interface{}{
+				"lastBroker": lastBrokerName,
+				"newBroker":  mq.BrokerName,
+			})
+		}
+
 		addr := p.client.GetNameSrv().FindBrokerAddrByName(mq.BrokerName)
 		if addr == "" {
 			return fmt.Errorf("topic=%s route info not found", mq.Topic)
@@ -333,6 +360,10 @@ func (p *defaultProducer) sendSync(ctx context.Context, msg *primitive.Message,
 			err = _err
 			continue
 		}
+
+		if needRetryCode(res.Code) && retryCount < retryTime-1 {
+			continue
+		}
 		return p.client.ProcessSendResponse(mq.BrokerName, res, resp, msg)
 	}
 	return err
@@ -359,7 +390,7 @@ func (p *defaultProducer) SendAsync(ctx context.Context, f func(context.Context,
 
 func (p *defaultProducer) sendAsync(ctx context.Context, msg *primitive.Message, h func(context.Context, *primitive.SendResult, error)) error {
 
-	mq := p.selectMessageQueue(msg)
+	mq := p.selectMessageQueue(msg, "")
 	if mq == nil {
 		return errors.Errorf("the topic=%s route info not found", msg.Topic)
 	}
@@ -416,8 +447,13 @@ func (p *defaultProducer) sendOneWay(ctx context.Context, msg *primitive.Message
 	retryTime := 1 + p.options.RetryTimes
 
 	var err error
+	var mq *primitive.MessageQueue
 	for retryCount := 0; retryCount < retryTime; retryCount++ {
-		mq := p.selectMessageQueue(msg)
+		var lastBrokerName string
+		if mq != nil {
+			lastBrokerName = mq.BrokerName
+		}
+		mq = p.selectMessageQueue(msg, lastBrokerName)
 		if mq == nil {
 			err = fmt.Errorf("the topic=%s route info not found", msg.Topic)
 			continue
@@ -554,13 +590,16 @@ func (p *defaultProducer) tryToFindTopicPublishInfo(topic string) *internal.Topi
 	return result
 }
 
-func (p *defaultProducer) selectMessageQueue(msg *primitive.Message) *primitive.MessageQueue {
-	topic := msg.Topic
-	result := p.tryToFindTopicPublishInfo(topic)
-	if result == nil {
+func (p *defaultProducer) selectMessageQueue(msg *primitive.Message, lastBrokerName string) *primitive.MessageQueue {
+	result := p.tryToFindTopicPublishInfo(msg.Topic)
+	if result == nil || len(result.MqList) == 0 {
+		rlog.Warning("topic route info is nil or empty", map[string]interface{}{
+			rlog.LogKeyTopic: msg.Topic,
+			"result":         result,
+		})
 		return nil
 	}
-	return p.options.Selector.Select(msg, result.MqList)
+	return p.options.Selector.Select(msg, result.MqList, lastBrokerName)
 }
 
 func (p *defaultProducer) PublishTopicList() []string {
diff --git a/producer/selector.go b/producer/selector.go
index 74f5bad..c9da49a 100644
--- a/producer/selector.go
+++ b/producer/selector.go
@@ -27,7 +27,7 @@ import (
 )
 
 type QueueSelector interface {
-	Select(*primitive.Message, []*primitive.MessageQueue) *primitive.MessageQueue
+	Select(msg *primitive.Message, mqs []*primitive.MessageQueue, lastBrokerName string) *primitive.MessageQueue
 }
 
 // manualQueueSelector use the queue manually set in the provided Message's QueueID  field as the queue to send.
@@ -37,7 +37,7 @@ func NewManualQueueSelector() QueueSelector {
 	return new(manualQueueSelector)
 }
 
-func (manualQueueSelector) Select(message *primitive.Message, queues []*primitive.MessageQueue) *primitive.MessageQueue {
+func (manualQueueSelector) Select(message *primitive.Message, queues []*primitive.MessageQueue, lastBrokerName string) *primitive.MessageQueue {
 	return message.Queue
 }
 
@@ -53,7 +53,7 @@ func NewRandomQueueSelector() QueueSelector {
 	return s
 }
 
-func (r *randomQueueSelector) Select(message *primitive.Message, queues []*primitive.MessageQueue) *primitive.MessageQueue {
+func (r *randomQueueSelector) Select(message *primitive.Message, queues []*primitive.MessageQueue, lastBrokerName string) *primitive.MessageQueue {
 	r.mux.Lock()
 	i := r.rander.Intn(len(queues))
 	r.mux.Unlock()
@@ -74,11 +74,32 @@ func NewRoundRobinQueueSelector() QueueSelector {
 	return s
 }
 
-func (r *roundRobinQueueSelector) Select(message *primitive.Message, queues []*primitive.MessageQueue) *primitive.MessageQueue {
+func (r *roundRobinQueueSelector) Select(message *primitive.Message, queues []*primitive.MessageQueue, lastBrokerName string) *primitive.MessageQueue {
 	t := message.Topic
-	var idx *uint32
 
 	r.Lock()
+	defer r.Unlock()
+	if lastBrokerName != "" {
+		for i := 0; i < len(queues); i++ {
+			idx, exist := r.indexer[t]
+			if !exist {
+				var v uint32 = 0
+				idx = &v
+				r.indexer[t] = idx
+			}
+			*idx++
+			qIndex := *idx % uint32(len(queues))
+			if queues[qIndex].BrokerName != lastBrokerName {
+				return queues[qIndex]
+			}
+		}
+	}
+	return r.selectOneMessageQueue(t, queues)
+}
+
+func (r *roundRobinQueueSelector) selectOneMessageQueue(t string, queues []*primitive.MessageQueue) *primitive.MessageQueue {
+	var idx *uint32
+
 	idx, exist := r.indexer[t]
 	if !exist {
 		var v uint32 = 0
@@ -86,7 +107,6 @@ func (r *roundRobinQueueSelector) Select(message *primitive.Message, queues []*p
 		r.indexer[t] = idx
 	}
 	*idx++
-	r.Unlock()
 
 	qIndex := *idx % uint32(len(queues))
 	return queues[qIndex]
@@ -103,10 +123,10 @@ func NewHashQueueSelector() QueueSelector {
 }
 
 // hashQueueSelector choose the queue by hash if message having sharding key, otherwise choose queue by random instead.
-func (h *hashQueueSelector) Select(message *primitive.Message, queues []*primitive.MessageQueue) *primitive.MessageQueue {
+func (h *hashQueueSelector) Select(message *primitive.Message, queues []*primitive.MessageQueue, lastBrokerName string) *primitive.MessageQueue {
 	key := message.GetShardingKey()
 	if len(key) == 0 {
-		return h.random.Select(message, queues)
+		return h.random.Select(message, queues, lastBrokerName)
 	}
 
 	hasher := fnv.New32a()


Test patch:
diff --git a/consumer/option_test.go b/consumer/option_test.go
index a1c016e..ab99b63 100644
--- a/consumer/option_test.go
+++ b/consumer/option_test.go
@@ -1,9 +1,7 @@
 package consumer
 
 import (
-	"fmt"
 	"reflect"
-	"strings"
 	"testing"
 )
 
@@ -15,7 +13,7 @@ func getFieldString(obj interface{}, field string) string {
 }
 
 func TestWithUnitName(t *testing.T) {
-	opt := defaultPullConsumerOptions()
+	opt := defaultPushConsumerOptions()
 	unitName := "unsh"
 	WithUnitName(unitName)(&opt)
 	if opt.UnitName != unitName {
@@ -24,7 +22,7 @@ func TestWithUnitName(t *testing.T) {
 }
 
 func TestWithNameServerDomain(t *testing.T) {
-	opt := defaultPullConsumerOptions()
+	opt := defaultPushConsumerOptions()
 	nameServerAddr := "http://127.0.0.1:8080/nameserver/addr"
 	WithNameServerDomain(nameServerAddr)(&opt)
 	domainStr := getFieldString(opt.Resolver, "domain")
@@ -34,30 +32,58 @@ func TestWithNameServerDomain(t *testing.T) {
 }
 
 func TestWithNameServerDomainAndUnitName(t *testing.T) {
-	nameServerAddr := "http://127.0.0.1:8080/nameserver/addr"
 	unitName := "unsh"
-	suffix := fmt.Sprintf("-%s?nofix=1", unitName)
-
 	// test with two different orders
 	t.Run("WithNameServerDomain & WithUnitName", func(t *testing.T) {
-		opt := defaultPullConsumerOptions()
-		WithNameServerDomain(nameServerAddr)(&opt)
+		addr := "http://127.0.0.1:8080/nameserver/addr"
+		opt := defaultPushConsumerOptions()
+		WithNameServerDomain(addr)(&opt)
+		WithUnitName(unitName)(&opt)
+
+		domainStr := getFieldString(opt.Resolver, "domain")
+		expectedAddr := "http://127.0.0.1:8080/nameserver/addr-unsh?nofix=1"
+		if domainStr != expectedAddr {
+			t.Errorf("consumer option WithNameServerDomain & WithUnitName. want:%s, got=%s", expectedAddr, domainStr)
+		}
+	})
+
+	t.Run("WithUnitName & WithNameServerDomain", func(t *testing.T) {
+		addr := "http://127.0.0.1:8080/nameserver/addr"
+		opt := defaultPushConsumerOptions()
+		WithUnitName(unitName)(&opt)
+		WithNameServerDomain(addr)(&opt)
+
+		domainStr := getFieldString(opt.Resolver, "domain")
+		expectedAddr := "http://127.0.0.1:8080/nameserver/addr-unsh?nofix=1"
+		if domainStr != expectedAddr {
+			t.Errorf("consumer option WithUnitName & WithNameServerDomain. want:%s, got=%s", expectedAddr, domainStr)
+		}
+	})
+
+	// test with two different orders - name server with query string
+	t.Run("WithNameServerDomain & WithUnitName", func(t *testing.T) {
+		addr := "http://127.0.0.1:8080/nameserver/addr?labels=abc"
+		opt := defaultPushConsumerOptions()
+		WithNameServerDomain(addr)(&opt)
 		WithUnitName(unitName)(&opt)
 
 		domainStr := getFieldString(opt.Resolver, "domain")
-		if !strings.Contains(domainStr, nameServerAddr) || !strings.Contains(domainStr, suffix) {
-			t.Errorf("consumer option should contains %s and %s", nameServerAddr, suffix)
+		expectedAddr := "http://127.0.0.1:8080/nameserver/addr-unsh?nofix=1&labels=abc"
+		if domainStr != expectedAddr {
+			t.Errorf("consumer option WithNameServerDomain & WithUnitName. want:%s, got=%s", expectedAddr, domainStr)
 		}
 	})
 
 	t.Run("WithUnitName & WithNameServerDomain", func(t *testing.T) {
-		opt := defaultPullConsumerOptions()
-		WithNameServerDomain(nameServerAddr)(&opt)
+		addr := "http://127.0.0.1:8080/nameserver/addr?labels=abc"
+		opt := defaultPushConsumerOptions()
 		WithUnitName(unitName)(&opt)
+		WithNameServerDomain(addr)(&opt)
 
 		domainStr := getFieldString(opt.Resolver, "domain")
-		if !strings.Contains(domainStr, nameServerAddr) || !strings.Contains(domainStr, suffix) {
-			t.Errorf("consumer option should contains %s and %s", nameServerAddr, suffix)
+		expectedAddr := "http://127.0.0.1:8080/nameserver/addr-unsh?nofix=1&labels=abc"
+		if domainStr != expectedAddr {
+			t.Errorf("consumer option WithUnitName & WithNameServerDomain. want:%s, got=%s", expectedAddr, domainStr)
 		}
 	})
 }
diff --git a/primitive/message_test.go b/primitive/message_test.go
index e47a298..7248df7 100644
--- a/primitive/message_test.go
+++ b/primitive/message_test.go
@@ -36,3 +36,13 @@ func TestMessageID(t *testing.T) {
 	}
 	t.Log(msgID)
 }
+
+func TestMessageKey(t *testing.T) {
+	msg := &Message{}
+	expected := "testKey"
+	msg.WithKeys([]string{expected})
+	actual := msg.GetKeys()
+	if actual != expected {
+		t.Fatalf("get message key error: expected is '%s', actual is '%s'", expected, actual)
+	}
+}
diff --git a/producer/option_test.go b/producer/option_test.go
index c074510..723da03 100644
--- a/producer/option_test.go
+++ b/producer/option_test.go
@@ -1,9 +1,7 @@
 package producer
 
 import (
-	"fmt"
 	"reflect"
-	"strings"
 	"testing"
 )
 
@@ -19,7 +17,7 @@ func TestWithUnitName(t *testing.T) {
 	unitName := "unsh"
 	WithUnitName(unitName)(&opt)
 	if opt.UnitName != unitName {
-		t.Errorf("consumer option WithUnitName. want:%s, got=%s", unitName, opt.UnitName)
+		t.Errorf("producer option WithUnitName. want:%s, got=%s", unitName, opt.UnitName)
 	}
 }
 
@@ -29,35 +27,63 @@ func TestWithNameServerDomain(t *testing.T) {
 	WithNameServerDomain(nameServerAddr)(&opt)
 	domainStr := getFieldString(opt.Resolver, "domain")
 	if domainStr != nameServerAddr {
-		t.Errorf("consumer option WithUnitName. want:%s, got=%s", nameServerAddr, domainStr)
+		t.Errorf("producer option WithUnitName. want:%s, got=%s", nameServerAddr, domainStr)
 	}
 }
 
 func TestWithNameServerDomainAndUnitName(t *testing.T) {
-	nameServerAddr := "http://127.0.0.1:8080/nameserver/addr"
 	unitName := "unsh"
-	suffix := fmt.Sprintf("-%s?nofix=1", unitName)
-
 	// test with two different orders
 	t.Run("WithNameServerDomain & WithUnitName", func(t *testing.T) {
+		addr := "http://127.0.0.1:8080/nameserver/addr"
+		opt := defaultProducerOptions()
+		WithNameServerDomain(addr)(&opt)
+		WithUnitName(unitName)(&opt)
+
+		domainStr := getFieldString(opt.Resolver, "domain")
+		expectedAddr := "http://127.0.0.1:8080/nameserver/addr-unsh?nofix=1"
+		if domainStr != expectedAddr {
+			t.Errorf("producer option WithNameServerDomain & WithUnitName. want:%s, got=%s", expectedAddr, domainStr)
+		}
+	})
+
+	t.Run("WithUnitName & WithNameServerDomain", func(t *testing.T) {
+		addr := "http://127.0.0.1:8080/nameserver/addr"
+		opt := defaultProducerOptions()
+		WithUnitName(unitName)(&opt)
+		WithNameServerDomain(addr)(&opt)
+
+		domainStr := getFieldString(opt.Resolver, "domain")
+		expectedAddr := "http://127.0.0.1:8080/nameserver/addr-unsh?nofix=1"
+		if domainStr != expectedAddr {
+			t.Errorf("producer option WithUnitName & WithNameServerDomain. want:%s, got=%s", expectedAddr, domainStr)
+		}
+	})
+
+	// test with two different orders - name server with query string
+	t.Run("WithNameServerDomain & WithUnitName", func(t *testing.T) {
+		addr := "http://127.0.0.1:8080/nameserver/addr?labels=abc"
 		opt := defaultProducerOptions()
-		WithNameServerDomain(nameServerAddr)(&opt)
+		WithNameServerDomain(addr)(&opt)
 		WithUnitName(unitName)(&opt)
 
 		domainStr := getFieldString(opt.Resolver, "domain")
-		if !strings.Contains(domainStr, nameServerAddr) || !strings.Contains(domainStr, suffix) {
-			t.Errorf("consumer option should contains %s and %s", nameServerAddr, suffix)
+		expectedAddr := "http://127.0.0.1:8080/nameserver/addr-unsh?nofix=1&labels=abc"
+		if domainStr != expectedAddr {
+			t.Errorf("producer option WithNameServerDomain & WithUnitName. want:%s, got=%s", expectedAddr, domainStr)
 		}
 	})
 
 	t.Run("WithUnitName & WithNameServerDomain", func(t *testing.T) {
+		addr := "http://127.0.0.1:8080/nameserver/addr?labels=abc"
 		opt := defaultProducerOptions()
-		WithNameServerDomain(nameServerAddr)(&opt)
 		WithUnitName(unitName)(&opt)
+		WithNameServerDomain(addr)(&opt)
 
 		domainStr := getFieldString(opt.Resolver, "domain")
-		if !strings.Contains(domainStr, nameServerAddr) || !strings.Contains(domainStr, suffix) {
-			t.Errorf("consumer option should contains %s and %s", nameServerAddr, suffix)
+		expectedAddr := "http://127.0.0.1:8080/nameserver/addr-unsh?nofix=1&labels=abc"
+		if domainStr != expectedAddr {
+			t.Errorf("producer option WithUnitName & WithNameServerDomain. want:%s, got=%s", expectedAddr, domainStr)
 		}
 	})
 }
diff --git a/producer/producer_test.go b/producer/producer_test.go
index e1d72dd..cbe7426 100644
--- a/producer/producer_test.go
+++ b/producer/producer_test.go
@@ -149,7 +149,11 @@ func TestSync(t *testing.T) {
 
 	mockB4Send(p)
 
-	client.EXPECT().InvokeSync(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, nil)
+	cmd := &remote.RemotingCommand{
+		Code: internal.ResSuccess,
+	}
+
+	client.EXPECT().InvokeSync(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(cmd, nil)
 	client.EXPECT().ProcessSendResponse(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Do(
 		func(brokerName string, cmd *remote.RemotingCommand, resp *primitive.SendResult, msgs ...*primitive.Message) {
 			resp.Status = expectedResp.Status
@@ -309,7 +313,11 @@ func TestSyncWithNamespace(t *testing.T) {
 
 	mockB4Send(p)
 
-	client.EXPECT().InvokeSync(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, nil)
+	cmd := &remote.RemotingCommand{
+		Code: internal.ResSuccess,
+	}
+
+	client.EXPECT().InvokeSync(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(cmd, nil)
 	client.EXPECT().ProcessSendResponse(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Do(
 		func(brokerName string, cmd *remote.RemotingCommand, resp *primitive.SendResult, msgs ...*primitive.Message) {
 			resp.Status = expectedResp.Status
diff --git a/producer/selector_test.go b/producer/selector_test.go
index 72dc469..e0b4384 100644
--- a/producer/selector_test.go
+++ b/producer/selector_test.go
@@ -26,7 +26,7 @@ import (
 )
 
 func TestRoundRobin(t *testing.T) {
-	queues := make([]*primitive.MessageQueue, 10)
+	queues := make([]*primitive.MessageQueue, 0)
 	for i := 0; i < 10; i++ {
 		queues = append(queues, &primitive.MessageQueue{
 			QueueId: i,
@@ -41,18 +41,43 @@ func TestRoundRobin(t *testing.T) {
 		Topic: "rr",
 	}
 	for i := 0; i < 100; i++ {
-		q := s.Select(m, queues)
+		q := s.Select(m, queues, "")
 		expected := (i + 1) % len(queues)
 		assert.Equal(t, queues[expected], q, "i: %d", i)
 
-		qrr := s.Select(mrr, queues)
+		qrr := s.Select(mrr, queues, "")
 		expected = (i + 1) % len(queues)
 		assert.Equal(t, queues[expected], qrr, "i: %d", i)
 	}
 }
 
+func TestRoundRobinRetry(t *testing.T) {
+	queues := make([]*primitive.MessageQueue, 0)
+	brokerA := "brokerA"
+	brokerB := "brokerB"
+	for i := 0; i < 5; i++ {
+		queues = append(queues, &primitive.MessageQueue{
+			QueueId:    i,
+			BrokerName: brokerA,
+		})
+		queues = append(queues, &primitive.MessageQueue{
+			QueueId:    i,
+			BrokerName: brokerB,
+		})
+	}
+	s := NewRoundRobinQueueSelector()
+
+	m := &primitive.Message{
+		Topic: "test",
+	}
+	for i := 0; i < 100; i++ {
+		q := s.Select(m, queues, brokerA)
+		assert.Equal(t, brokerB, q.BrokerName)
+	}
+}
+
 func TestHashQueueSelector(t *testing.T) {
-	queues := make([]*primitive.MessageQueue, 10)
+	queues := make([]*primitive.MessageQueue, 0)
 	for i := 0; i < 10; i++ {
 		queues = append(queues, &primitive.MessageQueue{
 			QueueId: i,
@@ -66,13 +91,13 @@ func TestHashQueueSelector(t *testing.T) {
 		Body:  []byte("one message"),
 	}
 	m1.WithShardingKey("same_key")
-	q1 := s.Select(m1, queues)
+	q1 := s.Select(m1, queues, "")
 
 	m2 := &primitive.Message{
 		Topic: "test",
 		Body:  []byte("another message"),
 	}
 	m2.WithShardingKey("same_key")
-	q2 := s.Select(m2, queues)
+	q2 := s.Select(m2, queues, "")
 	assert.Equal(t, *q1, *q2)
 }