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

Issue:
Faulty move of reference count of shared rmqClient
#888 change reference count of shared rmqClient into sync.Once body, which is buggy. The client will be shutdown once any consumer or producer call Shutdown, but it may be still used by other consumer or producer.
The problem mentioned in #888 will not happen, since the start of client is called in sync.Once: https://github.com/apache/rocketmq-client-go/blob/6c77b6792c224cff2f3f971fc44d4a7ab0d8a29e/consumer/push_consumer.go#L161
https://github.com/apache/rocketmq-client-go/blob/6c77b6792c224cff2f3f971fc44d4a7ab0d8a29e/consumer/consumer.go#L285, so it will only be called once. Multiple start call will not affect the reference count of client.
Simply revert #888 will fix the issue.

Pull request description:
[ISSUE #1018] Fix unintended shutdown of shared rmqClient
caused by missing ref count

## What is the purpose of the change

fix #1018 

## Brief changelog

revert #888 

## 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/README.md b/README.md
index 33a8d9d..f01c22c 100644
--- a/README.md
+++ b/README.md
@@ -32,7 +32,7 @@ For 2.X version, it supports:
 * consuming message using broadcast model
 * message tracing for pub and sub messages
 * ACL for producers and consumers
-* requset-reply model
+* request-reply model
 
 ----------
 ## How to use
diff --git a/consumer/consumer.go b/consumer/consumer.go
index 7bd2b02..acce804 100644
--- a/consumer/consumer.go
+++ b/consumer/consumer.go
@@ -430,6 +430,25 @@ func (dc *defaultConsumer) doBalance() {
 		}
 		return true
 	})
+	dc.truncateMessageQueueNotMyTopic()
+}
+
+func (dc *defaultConsumer) truncateMessageQueueNotMyTopic() {
+	dc.processQueueTable.Range(func(key, value interface{}) bool {
+		mq := key.(primitive.MessageQueue)
+		pq := value.(*processQueue)
+		if _, ok := dc.subscriptionDataTable.Load(mq.Topic); !ok {
+			pq.WithDropped(true)
+			if dc.removeUnnecessaryMessageQueue(&mq, pq) {
+				dc.processQueueTable.Delete(key)
+				rlog.Info("remove unnecessary mq because unsubscribed", map[string]interface{}{
+					rlog.LogKeyConsumerGroup: dc.consumerGroup,
+					rlog.LogKeyMessageQueue:  mq.String(),
+				})
+			}
+		}
+		return true
+	})
 }
 
 func (dc *defaultConsumer) SubscriptionDataList() []*internal.SubscriptionData {
diff --git a/consumer/option.go b/consumer/option.go
index f2cb9e8..3f7e6ca 100644
--- a/consumer/option.go
+++ b/consumer/option.go
@@ -18,9 +18,10 @@ limitations under the License.
 package consumer
 
 import (
-	"github.com/apache/rocketmq-client-go/v2/hooks"
+	"strings"
 	"time"
 
+	"github.com/apache/rocketmq-client-go/v2/hooks"
 	"github.com/apache/rocketmq-client-go/v2/internal"
 	"github.com/apache/rocketmq-client-go/v2/primitive"
 )
@@ -327,7 +328,21 @@ func WithNameServer(nameServers primitive.NamesrvAddr) Option {
 // WithNameServerDomain set NameServer domain
 func WithNameServerDomain(nameServerUrl string) Option {
 	return func(opts *consumerOptions) {
-		opts.Resolver = primitive.NewHttpResolver("DEFAULT", nameServerUrl)
+		h := primitive.NewHttpResolver("DEFAULT", nameServerUrl)
+		if opts.UnitName != "" {
+			h.DomainWithUnit(opts.UnitName)
+		}
+		opts.Resolver = h
+	}
+}
+
+// WithUnitName set the name of specified unit
+func WithUnitName(unitName string) Option {
+	return func(opts *consumerOptions) {
+		opts.UnitName = strings.TrimSpace(unitName)
+		if ns, ok := opts.Resolver.(*primitive.HttpResolver); ok {
+			ns.DomainWithUnit(opts.UnitName)
+		}
 	}
 }
 
diff --git a/consumer/process_queue.go b/consumer/process_queue.go
index aeb66d8..120595f 100644
--- a/consumer/process_queue.go
+++ b/consumer/process_queue.go
@@ -259,22 +259,32 @@ func (pq *processQueue) cleanExpiredMsg(pc *pushConsumer) {
 					"time":                   startTime,
 					rlog.LogKeyUnderlayError: err,
 				})
+				pq.mutex.RUnlock()
 				continue
 			}
 			if time.Now().UnixNano()/1e6-st <= int64(pc.option.ConsumeTimeout/time.Millisecond) {
 				pq.mutex.RUnlock()
 				return
 			}
-		}
-		pq.mutex.RUnlock()
-
-		if !pc.sendMessageBack("", msg, int(3+msg.ReconsumeTimes)) {
-			rlog.Error("send message back to broker error when clean expired messages", map[string]interface{}{
-				rlog.LogKeyConsumerGroup: pc.consumerGroup,
+			rlog.Info("send expire msg back. ", map[string]interface{}{
+				rlog.LogKeyTopic:       msg.Topic,
+				rlog.LogKeyMessageId:   msg.MsgId,
+				"startTime":            startTime,
+				rlog.LogKeyStoreHost:   msg.StoreHost,
+				rlog.LogKeyQueueId:     msg.Queue.QueueId,
+				rlog.LogKeyQueueOffset: msg.QueueOffset,
 			})
-			continue
+			pq.mutex.RUnlock()
+			if !pc.sendMessageBack("", msg, int(3+msg.ReconsumeTimes)) {
+				rlog.Error("send message back to broker error when clean expired messages", map[string]interface{}{
+					rlog.LogKeyConsumerGroup: pc.consumerGroup,
+				})
+				continue
+			}
+			pq.removeMessage(msg)
+		} else {
+			pq.mutex.RUnlock()
 		}
-		pq.removeMessage(msg)
 	}
 }
 
diff --git a/consumer/push_consumer.go b/consumer/push_consumer.go
index 85f9725..d8706b0 100644
--- a/consumer/push_consumer.go
+++ b/consumer/push_consumer.go
@@ -540,7 +540,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 {
@@ -1133,8 +1135,19 @@ func (pc *pushConsumer) consumeMessageConcurrently(pq *processQueue, mq *primiti
 
 			consumeRT := time.Now().Sub(beginTime)
 			if err != nil {
+				rlog.Warning("consumeMessageCurrently error", map[string]interface{}{
+					rlog.LogKeyUnderlayError: err,
+					rlog.LogKeyMessages:      msgs,
+					rlog.LogKeyMessageQueue:  mq,
+					rlog.LogKeyConsumerGroup: pc.consumerGroup,
+				})
 				msgCtx.Properties[primitive.PropCtxType] = string(primitive.ExceptionReturn)
 			} else if consumeRT >= pc.option.ConsumeTimeout {
+				rlog.Warning("consumeMessageCurrently time out", map[string]interface{}{
+					rlog.LogKeyMessages:      msgs,
+					rlog.LogKeyMessageQueue:  mq,
+					rlog.LogKeyConsumerGroup: pc.consumerGroup,
+				})
 				msgCtx.Properties[primitive.PropCtxType] = string(primitive.TimeoutReturn)
 			} else if result == ConsumeSuccess {
 				msgCtx.Properties[primitive.PropCtxType] = string(primitive.SuccessReturn)
@@ -1262,7 +1275,15 @@ func (pc *pushConsumer) consumeMessageOrderly(pq *processQueue, mq *primitive.Me
 			ctx = primitive.WithOrderlyCtx(ctx, orderlyCtx)
 
 			pq.lockConsume.Lock()
-			result, _ := pc.consumeInner(ctx, msgs)
+			result, err := pc.consumeInner(ctx, msgs)
+			if err != nil {
+				rlog.Warning("consumeMessage orderly error", map[string]interface{}{
+					rlog.LogKeyUnderlayError: err,
+					rlog.LogKeyMessages:      msgs,
+					rlog.LogKeyMessageQueue:  mq.String(),
+					rlog.LogKeyConsumerGroup: pc.consumerGroup,
+				})
+			}
 			pq.lockConsume.Unlock()
 
 			if result == Rollback || result == SuspendCurrentQueueAMoment {
diff --git a/internal/client.go b/internal/client.go
index f7b2aec..486dffd 100644
--- a/internal/client.go
+++ b/internal/client.go
@@ -19,7 +19,6 @@ package internal
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"net"
 	"os"
@@ -385,10 +384,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))
 		}
@@ -704,7 +701,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)
diff --git a/internal/model.go b/internal/model.go
index 0fd7955..f058925 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())
 		}
 	}
 
diff --git a/internal/trace.go b/internal/trace.go
index ad30281..f7cea8d 100644
--- a/internal/trace.go
+++ b/internal/trace.go
@@ -305,8 +305,11 @@ func (td *traceDispatcher) GetTraceTopicName() string {
 func (td *traceDispatcher) Start() {
 	td.running = true
 	td.cli.Start()
+	maxWaitDuration := 5 * time.Millisecond
+	td.ticker = time.NewTicker(maxWaitDuration)
+	maxWaitTime := maxWaitDuration.Nanoseconds()
 	go primitive.WithRecover(func() {
-		td.process()
+		td.process(maxWaitTime)
 	})
 }
 
@@ -334,12 +337,9 @@ func (td *traceDispatcher) Append(ctx TraceContext) bool {
 }
 
 // process
-func (td *traceDispatcher) process() {
+func (td *traceDispatcher) process(maxWaitTime int64) {
 	var count int
 	var batch []TraceContext
-	maxWaitDuration := 5 * time.Millisecond
-	maxWaitTime := maxWaitDuration.Nanoseconds()
-	td.ticker = time.NewTicker(maxWaitDuration)
 	lastput := time.Now()
 	for {
 		select {
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 4e5917c..970242f 100644
--- a/primitive/nsresolver.go
+++ b/primitive/nsresolver.go
@@ -6,7 +6,7 @@ The ASF licenses this file to You under the Apache License, Version 2.0
 (the "License"); you may not use this file except in compliance with
 the License.  You may obtain a copy of the License at
 
-    http://www.apache.org/licenses/LICENSE-2.0
+	http://www.apache.org/licenses/LICENSE-2.0
 
 Unless required by applicable law or agreed to in writing, software
 distributed under the License is distributed on an "AS IS" BASIS,
@@ -111,6 +111,16 @@ func NewHttpResolver(instance string, domain ...string) *HttpResolver {
 	return h
 }
 
+func (h *HttpResolver) DomainWithUnit(unitName string) {
+	if unitName == "" {
+		return
+	}
+	if strings.Contains(h.domain, "?nofix=1") {
+		return
+	}
+	h.domain = fmt.Sprintf("%s-%s?nofix=1", h.domain, unitName)
+}
+
 func (h *HttpResolver) Resolve() []string {
 	addrs := h.get()
 	if len(addrs) > 0 {
@@ -152,14 +162,14 @@ func (h *HttpResolver) get() []string {
 		return nil
 	}
 
-	bodyStr := string(body)
+	bodyStr := strings.TrimSpace(string(body))
 	if bodyStr == "" {
 		return nil
 	}
 
-	h.saveSnapshot(body)
+	_ = h.saveSnapshot([]byte(bodyStr))
 
-	return strings.Split(string(body), ";")
+	return strings.Split(bodyStr, ";")
 }
 
 func (h *HttpResolver) saveSnapshot(body []byte) error {
diff --git a/producer/option.go b/producer/option.go
index 9fd8374..c3a0dc4 100644
--- a/producer/option.go
+++ b/producer/option.go
@@ -18,6 +18,7 @@ limitations under the License.
 package producer
 
 import (
+	"strings"
 	"time"
 
 	"github.com/apache/rocketmq-client-go/v2/internal"
@@ -144,7 +145,21 @@ func WithNameServer(nameServers primitive.NamesrvAddr) Option {
 // WithNameServerDomain set NameServer domain
 func WithNameServerDomain(nameServerUrl string) Option {
 	return func(opts *producerOptions) {
-		opts.Resolver = primitive.NewHttpResolver("DEFAULT", nameServerUrl)
+		h := primitive.NewHttpResolver("DEFAULT", nameServerUrl)
+		if opts.UnitName != "" {
+			h.DomainWithUnit(opts.UnitName)
+		}
+		opts.Resolver = h
+	}
+}
+
+// WithUnitName set the name of specified unit
+func WithUnitName(unitName string) Option {
+	return func(opts *producerOptions) {
+		opts.UnitName = strings.TrimSpace(unitName)
+		if ns, ok := opts.Resolver.(*primitive.HttpResolver); ok {
+			ns.DomainWithUnit(opts.UnitName)
+		}
 	}
 }
 
diff --git a/producer/producer.go b/producer/producer.go
index 5586846..ebd5e33 100644
--- a/producer/producer.go
+++ b/producer/producer.go
@@ -492,16 +492,18 @@ func (p *defaultProducer) buildSendRequest(mq *primitive.MessageQueue,
 	}
 
 	req := &internal.SendMessageRequestHeader{
-		ProducerGroup:  p.group,
-		Topic:          mq.Topic,
-		QueueId:        mq.QueueId,
-		SysFlag:        sysFlag,
-		BornTimestamp:  time.Now().UnixNano() / int64(time.Millisecond),
-		Flag:           msg.Flag,
-		Properties:     msg.MarshallProperties(),
-		ReconsumeTimes: 0,
-		UnitMode:       p.options.UnitMode,
-		Batch:          msg.Batch,
+		ProducerGroup:         p.group,
+		Topic:                 mq.Topic,
+		QueueId:               mq.QueueId,
+		SysFlag:               sysFlag,
+		BornTimestamp:         time.Now().UnixNano() / int64(time.Millisecond),
+		Flag:                  msg.Flag,
+		Properties:            msg.MarshallProperties(),
+		ReconsumeTimes:        0,
+		UnitMode:              p.options.UnitMode,
+		Batch:                 msg.Batch,
+		DefaultTopic:          p.options.CreateTopicKey,
+		DefaultTopicQueueNums: p.options.DefaultTopicQueueNums,
 	}
 
 	msgType := msg.GetProperty(primitive.PropertyMsgType)
diff --git a/rlog/log.go b/rlog/log.go
index e4070b7..5c99e2d 100644
--- a/rlog/log.go
+++ b/rlog/log.go
@@ -39,6 +39,11 @@ const (
 	LogKeyValueChangedTo       = "changeTo"
 	LogKeyPullRequest          = "PullRequest"
 	LogKeyTimeStamp            = "timestamp"
+	LogKeyMessageId            = "msgId"
+	LogKeyStoreHost            = "storeHost"
+	LogKeyQueueId              = "queueId"
+	LogKeyQueueOffset          = "queueOffset"
+	LogKeyMessages             = "messages"
 )
 
 type Logger interface {


Test patch:
diff --git a/consumer/option_test.go b/consumer/option_test.go
new file mode 100644
index 0000000..a1c016e
--- /dev/null
+++ b/consumer/option_test.go
@@ -0,0 +1,63 @@
+package consumer
+
+import (
+	"fmt"
+	"reflect"
+	"strings"
+	"testing"
+)
+
+func getFieldString(obj interface{}, field string) string {
+	v := reflect.Indirect(reflect.ValueOf(obj))
+	return v.FieldByNameFunc(func(n string) bool {
+		return n == field
+	}).String()
+}
+
+func TestWithUnitName(t *testing.T) {
+	opt := defaultPullConsumerOptions()
+	unitName := "unsh"
+	WithUnitName(unitName)(&opt)
+	if opt.UnitName != unitName {
+		t.Errorf("consumer option WithUnitName. want:%s, got=%s", unitName, opt.UnitName)
+	}
+}
+
+func TestWithNameServerDomain(t *testing.T) {
+	opt := defaultPullConsumerOptions()
+	nameServerAddr := "http://127.0.0.1:8080/nameserver/addr"
+	WithNameServerDomain(nameServerAddr)(&opt)
+	domainStr := getFieldString(opt.Resolver, "domain")
+	if domainStr != nameServerAddr {
+		t.Errorf("consumer 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) {
+		opt := defaultPullConsumerOptions()
+		WithNameServerDomain(nameServerAddr)(&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)
+		}
+	})
+
+	t.Run("WithUnitName & WithNameServerDomain", func(t *testing.T) {
+		opt := defaultPullConsumerOptions()
+		WithNameServerDomain(nameServerAddr)(&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)
+		}
+	})
+}
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/primitive/nsresolver_test.go b/primitive/nsresolver_test.go
index d42d2c6..94d80db 100644
--- a/primitive/nsresolver_test.go
+++ b/primitive/nsresolver_test.go
@@ -6,7 +6,7 @@ The ASF licenses this file to You under the Apache License, Version 2.0
 (the "License"); you may not use this file except in compliance with
 the License.  You may obtain a copy of the License at
 
-    http://www.apache.org/licenses/LICENSE-2.0
+	http://www.apache.org/licenses/LICENSE-2.0
 
 Unless required by applicable law or agreed to in writing, software
 distributed under the License is distributed on an "AS IS" BASIS,
@@ -18,7 +18,6 @@ package primitive
 
 import (
 	"fmt"
-	"github.com/apache/rocketmq-client-go/v2/rlog"
 	"io/ioutil"
 	"net"
 	"net/http"
@@ -26,6 +25,8 @@ import (
 	"strings"
 	"testing"
 
+	"github.com/apache/rocketmq-client-go/v2/rlog"
+
 	. "github.com/smartystreets/goconvey/convey"
 )
 
@@ -81,6 +82,43 @@ func TestHttpResolverWithGet(t *testing.T) {
 	})
 }
 
+func TestHttpResolverWithGetUnitName(t *testing.T) {
+	Convey("Test UpdateNameServerAddress Save Local Snapshot", t, func() {
+		srvs := []string{
+			"192.168.100.1",
+			"192.168.100.2",
+			"192.168.100.3",
+			"192.168.100.4",
+			"192.168.100.5",
+		}
+		http.HandleFunc("/nameserver/addrs3-unsh", func(w http.ResponseWriter, r *http.Request) {
+			if r.URL.Query().Get("nofix") == "1" {
+				fmt.Fprintf(w, strings.Join(srvs, ";"))
+			}
+			fmt.Fprintf(w, "")
+		})
+		server := &http.Server{Addr: ":0", Handler: nil}
+		listener, _ := net.Listen("tcp", ":0")
+		go server.Serve(listener)
+
+		port := listener.Addr().(*net.TCPAddr).Port
+		nameServerDommain := fmt.Sprintf("http://127.0.0.1:%d/nameserver/addrs3", port)
+		rlog.Info("Temporary Nameserver", map[string]interface{}{
+			"domain": nameServerDommain,
+		})
+
+		resolver := NewHttpResolver("DEFAULT", nameServerDommain)
+		resolver.DomainWithUnit("unsh")
+		resolver.Resolve()
+
+		// check snapshot saved
+		filePath := resolver.getSnapshotFilePath("DEFAULT")
+		body := strings.Join(srvs, ";")
+		bs, _ := ioutil.ReadFile(filePath)
+		So(string(bs), ShouldEqual, body)
+	})
+}
+
 func TestHttpResolverWithSnapshotFile(t *testing.T) {
 	Convey("Test UpdateNameServerAddress Use Local Snapshot", t, func() {
 		srvs := []string{
diff --git a/producer/option_test.go b/producer/option_test.go
new file mode 100644
index 0000000..c074510
--- /dev/null
+++ b/producer/option_test.go
@@ -0,0 +1,63 @@
+package producer
+
+import (
+	"fmt"
+	"reflect"
+	"strings"
+	"testing"
+)
+
+func getFieldString(obj interface{}, field string) string {
+	v := reflect.Indirect(reflect.ValueOf(obj))
+	return v.FieldByNameFunc(func(n string) bool {
+		return n == field
+	}).String()
+}
+
+func TestWithUnitName(t *testing.T) {
+	opt := defaultProducerOptions()
+	unitName := "unsh"
+	WithUnitName(unitName)(&opt)
+	if opt.UnitName != unitName {
+		t.Errorf("consumer option WithUnitName. want:%s, got=%s", unitName, opt.UnitName)
+	}
+}
+
+func TestWithNameServerDomain(t *testing.T) {
+	opt := defaultProducerOptions()
+	nameServerAddr := "http://127.0.0.1:8080/nameserver/addr"
+	WithNameServerDomain(nameServerAddr)(&opt)
+	domainStr := getFieldString(opt.Resolver, "domain")
+	if domainStr != nameServerAddr {
+		t.Errorf("consumer 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) {
+		opt := defaultProducerOptions()
+		WithNameServerDomain(nameServerAddr)(&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)
+		}
+	})
+
+	t.Run("WithUnitName & WithNameServerDomain", func(t *testing.T) {
+		opt := defaultProducerOptions()
+		WithNameServerDomain(nameServerAddr)(&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)
+		}
+	})
+}