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

Issue:
The process queues of unsubscribed topic are not dropped
**BUG REPORT**  
1. Unsubscribed topic still has process queues running, which should be dropped during rebalance. The problem is caused by missing function implement compared to Java version:

https://github.com/apache/rocketmq/blob/06f2208a34907211591114f6b0d327168c250fb3/client/src/main/java/org/apache/rocketmq/client/impl/consumer/RebalanceImpl.java#L232

https://github.com/apache/rocketmq/blob/06f2208a34907211591114f6b0d327168c250fb3/client/src/main/java/org/apache/rocketmq/client/impl/consumer/RebalanceImpl.java#L314


Pull request description:
[ISSUE #1024] fix: unsubscribed topic queues not dropped
## What is the purpose of the change

fix #1024 

## 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/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/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/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)
+		}
 	}
 }
 


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/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)
+		}
+	})
+}