Repository: apache/rocketmq-client-go
Pull request #550 (2020-11-02 10:22:11)

[ISSUE #544] Fix bug, concurrent map read and map write local offest file
## What is the purpose of the change

 avoid `fatal error: concurrent map read and map write`

## Brief changelog

use sync.Map to save` localFileOffsetStore.OffsetTable` info.

## Verifying this change

`go test . -v --run TestNewLocalFileOffsetStore` pass
`go test . -v --run TestLocalFileOffsetStore` pass



Patch:
diff --git a/.gitignore b/.gitignore
index a0c292d..99fd04d 100644
--- a/.gitignore
+++ b/.gitignore
@@ -3,4 +3,5 @@ go.mod
 go.sum
 vendor/
 coverage.txt
-examples/test
\ No newline at end of file
+examples/test
+/.vscode
\ No newline at end of file
diff --git a/LICENSE b/LICENSE
index 0130891..e6e77b0 100644
--- a/LICENSE
+++ b/LICENSE
@@ -15,7 +15,7 @@
       "Legal Entity" shall mean the union of the acting entity and all
       other entities that control, are controlled by, or are under common
       control with that entity. For the purposes of this definition,
-      "control" means (properties) the power, direct or indirect, to cause the
+      "control" means (i) the power, direct or indirect, to cause the
       direction or management of such entity, whether by contract or
       otherwise, or (ii) ownership of fifty percent (50%) or more of the
       outstanding shares, or (iii) beneficial ownership of such entity.
@@ -198,4 +198,4 @@
    distributed under the License is distributed on an "AS IS" BASIS,
    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    See the License for the specific language governing permissions and
-   limitations under the License.
\ No newline at end of file
+   limitations under the License.
diff --git a/README.md b/README.md
index 477a590..08494a6 100644
--- a/README.md
+++ b/README.md
@@ -7,7 +7,7 @@
 [![GitHub release](https://img.shields.io/badge/release-download-default.svg)](https://github.com/apache/rocketmq-client-go/releases)
 [![Average time to resolve an issue](http://isitmaintained.com/badge/resolution/apache/rocketmq-client-go.svg)](http://isitmaintained.com/project/apache/rocketmq-client-go "Average time to resolve an issue")
 [![Percentage of issues still open](http://isitmaintained.com/badge/open/apache/rocketmq-client-go.svg)](http://isitmaintained.com/project/apache/rocketmq-client-go "Percentage of issues still open")
-![Twitter Follow](https://img.shields.io/twitter/follow/ApacheRocketMQ?style=social)
+[![Twitter Follow](https://img.shields.io/twitter/follow/ApacheRocketMQ?style=social)](https://twitter.com/intent/follow?screen_name=ApacheRocketMQ)
 
 A product ready RocketMQ Client in pure go, which supports almost the full features of Apache RocketMQ, such as pub and sub messages, ACL, tracing and so on.
 
@@ -52,4 +52,4 @@ For 2.0.0 version, it supports:
    
 ----------
 ## License
-  [Apache License, Version 2.0](http://www.apache.org/licenses/LICENSE-2.0.html) Copyright (C) Apache Software Foundation
+  [Apache License, Version 2.0](http://www.apache.org/licenses/LICENSE-2.0.html) Copyright (C) Apache Software Foundation
\ No newline at end of file
diff --git a/consumer/offset_store.go b/consumer/offset_store.go
index 17f5d76..4d0b271 100644
--- a/consumer/offset_store.go
+++ b/consumer/offset_store.go
@@ -26,13 +26,12 @@ import (
 	"sync"
 	"time"
 
-	jsoniter "github.com/json-iterator/go"
-
 	"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"
 	"github.com/apache/rocketmq-client-go/v2/primitive"
 	"github.com/apache/rocketmq-client-go/v2/rlog"
+	jsoniter "github.com/json-iterator/go"
 )
 
 type readType int
@@ -101,7 +100,7 @@ func (mq *MessageQueueKey) UnmarshalText(text []byte) error {
 type localFileOffsetStore struct {
 	group       string
 	path        string
-	OffsetTable map[MessageQueueKey]int64
+	OffsetTable *sync.Map // concurrent safe , map[MessageQueueKey]int64
 	// mutex for offset file
 	mutex sync.Mutex
 }
@@ -110,7 +109,7 @@ func NewLocalFileOffsetStore(clientID, group string) OffsetStore {
 	store := &localFileOffsetStore{
 		group:       group,
 		path:        filepath.Join(_LocalOffsetStorePath, clientID, group, "offset.json"),
-		OffsetTable: make(map[MessageQueueKey]int64),
+		OffsetTable: new(sync.Map),
 	}
 	store.load()
 	return store
@@ -151,7 +150,9 @@ func (local *localFileOffsetStore) load() {
 	}
 
 	if datas != nil {
-		local.OffsetTable = datas
+		for k, v := range datas {
+			local.OffsetTable.Store(k, v)
+		}
 	}
 }
 
@@ -180,17 +181,17 @@ func (local *localFileOffsetStore) update(mq *primitive.MessageQueue, offset int
 		"new_offset":            offset,
 	})
 	key := MessageQueueKey(*mq)
-	localOffset, exist := local.OffsetTable[key]
+	localOffset, exist := local.OffsetTable.Load(key)
 	if !exist {
-		local.OffsetTable[key] = offset
+		local.OffsetTable.Store(key, offset)
 		return
 	}
 	if increaseOnly {
-		if localOffset < offset {
-			local.OffsetTable[key] = offset
+		if localOffset.(int64) < offset {
+			local.OffsetTable.Store(key, offset)
 		}
 	} else {
-		local.OffsetTable[key] = offset
+		local.OffsetTable.Store(key, offset)
 	}
 }
 
@@ -201,10 +202,17 @@ func (local *localFileOffsetStore) persist(mqs []*primitive.MessageQueue) {
 	local.mutex.Lock()
 	defer local.mutex.Unlock()
 
+	datas := make(map[MessageQueueKey]int64)
+	local.OffsetTable.Range(func(key, value interface{}) bool {
+		k := key.(MessageQueueKey)
+		v := value.(int64)
+		datas[k] = v
+		return true
+	})
+
 	wrapper := OffsetSerializeWrapper{
-		OffsetTable: local.OffsetTable,
+		OffsetTable: datas,
 	}
-
 	data, _ := jsoniter.Marshal(wrapper)
 	utils.CheckError(fmt.Sprintf("persist offset to %s", local.path), utils.WriteToFile(local.path, data))
 }
@@ -384,11 +392,11 @@ func (r *remoteBrokerOffsetStore) updateConsumeOffsetToBroker(group string, mq p
 	return r.client.InvokeOneWay(context.Background(), broker, cmd, 5*time.Second)
 }
 
-func readFromMemory(table map[MessageQueueKey]int64, mq *primitive.MessageQueue) int64 {
-	localOffset, exist := table[MessageQueueKey(*mq)]
+func readFromMemory(table *sync.Map, mq *primitive.MessageQueue) int64 {
+	localOffset, exist := table.Load(MessageQueueKey(*mq))
 	if !exist {
 		return -1
 	}
 
-	return localOffset
+	return localOffset.(int64)
 }
diff --git a/consumer/option.go b/consumer/option.go
index 07b4246..330f6c4 100644
--- a/consumer/option.go
+++ b/consumer/option.go
@@ -254,8 +254,8 @@ func WithPullInterval(interval time.Duration) Option {
 	}
 }
 
-// WithNsResovler set nameserver resolver to fetch nameserver addr
-func WithNsResovler(resolver primitive.NsResolver) Option {
+// WithNsResolver set nameserver resolver to fetch nameserver addr
+func WithNsResolver(resolver primitive.NsResolver) Option {
 	return func(options *consumerOptions) {
 		options.Resolver = resolver
 	}
diff --git a/consumer/push_consumer.go b/consumer/push_consumer.go
index 58945c2..5177fd9 100644
--- a/consumer/push_consumer.go
+++ b/consumer/push_consumer.go
@@ -222,9 +222,11 @@ func (pc *pushConsumer) Shutdown() error {
 
 func (pc *pushConsumer) Subscribe(topic string, selector MessageSelector,
 	f func(context.Context, ...*primitive.MessageExt) (ConsumeResult, error)) error {
-	if atomic.LoadInt32(&pc.state) != int32(internal.StateCreateJust) {
-		return errors.New("subscribe topic only started before")
+	if atomic.LoadInt32(&pc.state) == int32(internal.StateStartFailed) ||
+		atomic.LoadInt32(&pc.state) == int32(internal.StateShutdown) {
+		return errors.New("cannot subscribe topic since client either failed to start or has been shutdown.")
 	}
+
 	if pc.option.Namespace != "" {
 		topic = pc.option.Namespace + "%" + topic
 	}
@@ -514,7 +516,7 @@ func (pc *pushConsumer) pullMessage(request *PullRequest) {
 					"PullThresholdForQueue": pc.option.PullThresholdForQueue,
 					"minOffset":             pq.Min(),
 					"maxOffset":             pq.Max(),
-					"count":                 pq.msgCache,
+					"count":                 pq.cachedMsgCount,
 					"size(MiB)":             cachedMessageSizeInMiB,
 					"flowControlTimes":      pc.queueFlowControlTimes,
 					rlog.LogKeyPullRequest:  request.String(),
@@ -531,7 +533,7 @@ func (pc *pushConsumer) pullMessage(request *PullRequest) {
 					"PullThresholdSizeForQueue": pc.option.PullThresholdSizeForQueue,
 					"minOffset":                 pq.Min(),
 					"maxOffset":                 pq.Max(),
-					"count":                     pq.msgCache,
+					"count":                     pq.cachedMsgCount,
 					"size(MiB)":                 cachedMessageSizeInMiB,
 					"flowControlTimes":          pc.queueFlowControlTimes,
 					rlog.LogKeyPullRequest:      request.String(),
diff --git a/docs/Introduction.md b/docs/Introduction.md
index 2132b91..a4e954e 100644
--- a/docs/Introduction.md
+++ b/docs/Introduction.md
@@ -30,7 +30,7 @@ Producer interface {
 ```
 p, err := rocketmq.NewProducer(
 		producer.WithNameServer(endPoint),
-		//producer.WithNsResovler(primitive.NewPassthroughResolver(endPoint)),
+		//producer.WithNsResolver(primitive.NewPassthroughResolver(endPoint)),
 		producer.WithRetry(2),
 		producer.WithGroupName("GID_xxxxxx"),
 	)
diff --git a/examples/consumer/acl/main.go b/examples/consumer/acl/main.go
index a6535fd..96d0dd0 100644
--- a/examples/consumer/acl/main.go
+++ b/examples/consumer/acl/main.go
@@ -31,7 +31,7 @@ import (
 func main() {
 	c, err := rocketmq.NewPushConsumer(
 		consumer.WithGroupName("testGroup"),
-		consumer.WithNsResovler(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
+		consumer.WithNsResolver(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
 		consumer.WithCredentials(primitive.Credentials{
 			AccessKey: "RocketMQ",
 			SecretKey: "12345678",
diff --git a/examples/consumer/broadcast/main.go b/examples/consumer/broadcast/main.go
index 29b0b12..219ab3e 100644
--- a/examples/consumer/broadcast/main.go
+++ b/examples/consumer/broadcast/main.go
@@ -31,7 +31,7 @@ import (
 func main() {
 	c, _ := rocketmq.NewPushConsumer(
 		consumer.WithGroupName("testGroup"),
-		consumer.WithNsResovler(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
+		consumer.WithNsResolver(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
 		consumer.WithConsumeFromWhere(consumer.ConsumeFromFirstOffset),
 		consumer.WithConsumerModel(consumer.BroadCasting),
 	)
diff --git a/examples/consumer/delay/main.go b/examples/consumer/delay/main.go
index 8cc5c04..1bbe6ed 100644
--- a/examples/consumer/delay/main.go
+++ b/examples/consumer/delay/main.go
@@ -31,7 +31,7 @@ import (
 func main() {
 	c, _ := rocketmq.NewPushConsumer(
 		consumer.WithGroupName("testGroup"),
-		consumer.WithNsResovler(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
+		consumer.WithNsResolver(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
 	)
 	err := c.Subscribe("TopicTest", consumer.MessageSelector{}, func(ctx context.Context,
 		msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
diff --git a/examples/consumer/interceptor/main.go b/examples/consumer/interceptor/main.go
index 1036c6a..83f43b0 100644
--- a/examples/consumer/interceptor/main.go
+++ b/examples/consumer/interceptor/main.go
@@ -31,7 +31,7 @@ import (
 func main() {
 	c, _ := rocketmq.NewPushConsumer(
 		consumer.WithGroupName("testGroup"),
-		consumer.WithNsResovler(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
+		consumer.WithNsResolver(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
 		consumer.WithConsumerModel(consumer.Clustering),
 		consumer.WithConsumeFromWhere(consumer.ConsumeFromFirstOffset),
 		consumer.WithInterceptor(UserFistInterceptor(), UserSecondInterceptor()))
diff --git a/examples/consumer/namespace/main.go b/examples/consumer/namespace/main.go
index e1b9dea..d46f210 100644
--- a/examples/consumer/namespace/main.go
+++ b/examples/consumer/namespace/main.go
@@ -31,7 +31,7 @@ import (
 func main() {
 	c, err := rocketmq.NewPushConsumer(
 		consumer.WithGroupName("testGroup"),
-		consumer.WithNsResovler(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
+		consumer.WithNsResolver(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
 		consumer.WithCredentials(primitive.Credentials{
 			AccessKey: "RocketMQ",
 			SecretKey: "12345678",
diff --git a/examples/consumer/orderly/main.go b/examples/consumer/orderly/main.go
index 9e7a810..0f33e5f 100644
--- a/examples/consumer/orderly/main.go
+++ b/examples/consumer/orderly/main.go
@@ -31,7 +31,7 @@ import (
 func main() {
 	c, _ := rocketmq.NewPushConsumer(
 		consumer.WithGroupName("testGroup"),
-		consumer.WithNsResovler(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
+		consumer.WithNsResolver(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
 		consumer.WithConsumerModel(consumer.Clustering),
 		consumer.WithConsumeFromWhere(consumer.ConsumeFromFirstOffset),
 		consumer.WithConsumerOrder(true),
diff --git a/examples/consumer/pull/main.go b/examples/consumer/pull/main.go
index d740b15..3d49e7e 100644
--- a/examples/consumer/pull/main.go
+++ b/examples/consumer/pull/main.go
@@ -31,7 +31,7 @@ import (
 func main() {
 	c, err := rocketmq.NewPullConsumer(
 		consumer.WithGroupName("testGroup"),
-		consumer.WithNsResovler(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
+		consumer.WithNsResolver(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
 	)
 	if err != nil {
 		rlog.Fatal(fmt.Sprintf("fail to new pullConsumer: %s", err), nil)
diff --git a/examples/consumer/retry/concurrent/main.go b/examples/consumer/retry/concurrent/main.go
index 5fcf489..49f16d1 100644
--- a/examples/consumer/retry/concurrent/main.go
+++ b/examples/consumer/retry/concurrent/main.go
@@ -37,7 +37,7 @@ import (
 func main() {
 	c, _ := rocketmq.NewPushConsumer(
 		consumer.WithGroupName("testGroup"),
-		consumer.WithNsResovler(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
+		consumer.WithNsResolver(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
 		consumer.WithConsumerModel(consumer.Clustering),
 	)
 
diff --git a/examples/consumer/retry/order/main.go b/examples/consumer/retry/order/main.go
index 4ec05e7..e2d13bf 100644
--- a/examples/consumer/retry/order/main.go
+++ b/examples/consumer/retry/order/main.go
@@ -36,7 +36,7 @@ import (
 func main() {
 	c, _ := rocketmq.NewPushConsumer(
 		consumer.WithGroupName("testGroup"),
-		consumer.WithNsResovler(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
+		consumer.WithNsResolver(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
 		consumer.WithConsumerModel(consumer.Clustering),
 		consumer.WithConsumeFromWhere(consumer.ConsumeFromFirstOffset),
 		consumer.WithConsumerOrder(true),
diff --git a/examples/consumer/simple/main.go b/examples/consumer/simple/main.go
index 7d1a0b7..70f35b3 100644
--- a/examples/consumer/simple/main.go
+++ b/examples/consumer/simple/main.go
@@ -31,7 +31,7 @@ import (
 func main() {
 	c, _ := rocketmq.NewPushConsumer(
 		consumer.WithGroupName("testGroup"),
-		consumer.WithNsResovler(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
+		consumer.WithNsResolver(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
 	)
 	err := c.Subscribe("test", consumer.MessageSelector{}, func(ctx context.Context,
 		msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
diff --git a/examples/consumer/strategy/main.go b/examples/consumer/strategy/main.go
index 502524c..269ce11 100644
--- a/examples/consumer/strategy/main.go
+++ b/examples/consumer/strategy/main.go
@@ -31,7 +31,7 @@ import (
 func main() {
 	c, _ := rocketmq.NewPushConsumer(
 		consumer.WithGroupName("testGroup"),
-		consumer.WithNsResovler(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
+		consumer.WithNsResolver(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
 		consumer.WithStrategy(consumer.AllocateByAveragely),
 	)
 	err := c.Subscribe("TopicTest", consumer.MessageSelector{}, func(ctx context.Context,
diff --git a/examples/consumer/tag/main.go b/examples/consumer/tag/main.go
index 0532971..ec16f51 100644
--- a/examples/consumer/tag/main.go
+++ b/examples/consumer/tag/main.go
@@ -31,7 +31,7 @@ import (
 func main() {
 	c, _ := rocketmq.NewPushConsumer(
 		consumer.WithGroupName("testGroup"),
-		consumer.WithNsResovler(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
+		consumer.WithNsResolver(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
 	)
 	selector := consumer.MessageSelector{
 		Type:       consumer.TAG,
diff --git a/examples/consumer/trace/main.go b/examples/consumer/trace/main.go
index 4daaa9f..97f22ed 100644
--- a/examples/consumer/trace/main.go
+++ b/examples/consumer/trace/main.go
@@ -37,7 +37,7 @@ func main() {
 
 	c, _ := rocketmq.NewPushConsumer(
 		consumer.WithGroupName("testGroup"),
-		consumer.WithNsResovler(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
+		consumer.WithNsResolver(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
 		consumer.WithTrace(traceCfg),
 	)
 	err := c.Subscribe("TopicTest", consumer.MessageSelector{}, func(ctx context.Context,
diff --git a/examples/producer/acl/main.go b/examples/producer/acl/main.go
index 38d61dc..96881d3 100644
--- a/examples/producer/acl/main.go
+++ b/examples/producer/acl/main.go
@@ -30,7 +30,7 @@ import (
 
 func main() {
 	p, err := rocketmq.NewProducer(
-		producer.WithNsResovler(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
+		producer.WithNsResolver(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
 		producer.WithRetry(2),
 		producer.WithCredentials(primitive.Credentials{
 			AccessKey: "RocketMQ",
diff --git a/examples/producer/async/main.go b/examples/producer/async/main.go
index aa73881..2a5182c 100644
--- a/examples/producer/async/main.go
+++ b/examples/producer/async/main.go
@@ -31,7 +31,7 @@ import (
 // Package main implements a async producer to send message.
 func main() {
 	p, _ := rocketmq.NewProducer(
-		producer.WithNsResovler(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
+		producer.WithNsResolver(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
 		producer.WithRetry(2),
 		producer.WithQueueSelector(producer.NewManualQueueSelector()))
 
diff --git a/examples/producer/batch/main.go b/examples/producer/batch/main.go
index d807daf..dc591a1 100644
--- a/examples/producer/batch/main.go
+++ b/examples/producer/batch/main.go
@@ -30,7 +30,7 @@ import (
 
 func main() {
 	p, _ := rocketmq.NewProducer(
-		producer.WithNsResovler(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
+		producer.WithNsResolver(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
 		producer.WithRetry(2),
 	)
 	err := p.Start()
diff --git a/examples/producer/delay/main.go b/examples/producer/delay/main.go
index 465e97c..aadbb70 100644
--- a/examples/producer/delay/main.go
+++ b/examples/producer/delay/main.go
@@ -29,7 +29,7 @@ import (
 
 func main() {
 	p, _ := rocketmq.NewProducer(
-		producer.WithNsResovler(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
+		producer.WithNsResolver(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
 		producer.WithRetry(2),
 	)
 	err := p.Start()
diff --git a/examples/producer/interceptor/main.go b/examples/producer/interceptor/main.go
index 47ea1fb..c40f9a0 100644
--- a/examples/producer/interceptor/main.go
+++ b/examples/producer/interceptor/main.go
@@ -30,7 +30,7 @@ import (
 
 func main() {
 	p, _ := rocketmq.NewProducer(
-		producer.WithNsResovler(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
+		producer.WithNsResolver(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
 		producer.WithRetry(2),
 		producer.WithInterceptor(UserFirstInterceptor(), UserSecondInterceptor()),
 	)
diff --git a/examples/producer/namespace/main.go b/examples/producer/namespace/main.go
index 524bd32..9124284 100644
--- a/examples/producer/namespace/main.go
+++ b/examples/producer/namespace/main.go
@@ -30,7 +30,7 @@ import (
 
 func main() {
 	p, err := rocketmq.NewProducer(
-		producer.WithNsResovler(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
+		producer.WithNsResolver(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
 		producer.WithRetry(2),
 		producer.WithCredentials(primitive.Credentials{
 			AccessKey: "RocketMQ",
diff --git a/examples/producer/simple/main.go b/examples/producer/simple/main.go
index 8ac4421..6fd3364 100644
--- a/examples/producer/simple/main.go
+++ b/examples/producer/simple/main.go
@@ -31,7 +31,7 @@ import (
 // Package main implements a simple producer to send message.
 func main() {
 	p, _ := rocketmq.NewProducer(
-		producer.WithNsResovler(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
+		producer.WithNsResolver(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
 		producer.WithRetry(2),
 	)
 	err := p.Start()
diff --git a/examples/producer/tag/main.go b/examples/producer/tag/main.go
index 2bcd51b..7ce8559 100644
--- a/examples/producer/tag/main.go
+++ b/examples/producer/tag/main.go
@@ -29,7 +29,7 @@ import (
 
 func main() {
 	p, _ := rocketmq.NewProducer(
-		producer.WithNsResovler(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
+		producer.WithNsResolver(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
 		producer.WithRetry(2),
 	)
 	err := p.Start()
diff --git a/examples/producer/trace/main.go b/examples/producer/trace/main.go
index e6a25a9..b741704 100644
--- a/examples/producer/trace/main.go
+++ b/examples/producer/trace/main.go
@@ -36,7 +36,7 @@ func main() {
 	}
 
 	p, _ := rocketmq.NewProducer(
-		producer.WithNsResovler(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
+		producer.WithNsResolver(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
 		producer.WithRetry(2),
 		producer.WithTrace(traceCfg))
 	err := p.Start()
diff --git a/examples/producer/transaction/main.go b/examples/producer/transaction/main.go
index dde39a9..05b6c52 100644
--- a/examples/producer/transaction/main.go
+++ b/examples/producer/transaction/main.go
@@ -79,7 +79,7 @@ func (dl *DemoListener) CheckLocalTransaction(msg *primitive.MessageExt) primiti
 func main() {
 	p, _ := rocketmq.NewTransactionProducer(
 		NewDemoListener(),
-		producer.WithNsResovler(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
+		producer.WithNsResolver(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
 		producer.WithRetry(1),
 	)
 	err := p.Start()
diff --git a/internal/remote/remote_client.go b/internal/remote/remote_client.go
index 7b5b1ea..dae364e 100644
--- a/internal/remote/remote_client.go
+++ b/internal/remote/remote_client.go
@@ -194,7 +194,7 @@ func (c *remotingClient) processCMD(cmd *RemotingCommand, r *tcpConnWrapper) {
 				responseFuture.ResponseCommand = cmd
 				responseFuture.executeInvokeCallback()
 				if responseFuture.Done != nil {
-					responseFuture.Done <- true
+					close(responseFuture.Done)
 				}
 			})
 		}
diff --git a/primitive/nsresolver.go b/primitive/nsresolver.go
index d844373..4e5917c 100644
--- a/primitive/nsresolver.go
+++ b/primitive/nsresolver.go
@@ -130,11 +130,15 @@ func (h *HttpResolver) Description() string {
 
 func (h *HttpResolver) get() []string {
 	resp, err := h.cli.Get(h.domain)
-	if err != nil {
-		rlog.Error("name server http fetch failed", map[string]interface{}{
+	if err != nil || resp == nil || resp.StatusCode != 200 {
+		data := map[string]interface{}{
 			"NameServerDomain": h.domain,
 			"err":              err,
-		})
+		}
+		if resp != nil {
+			data["StatusCode"] = resp.StatusCode
+		}
+		rlog.Error("name server http fetch failed", data)
 		return nil
 	}
 
diff --git a/producer/option.go b/producer/option.go
index 76e9a31..bacef7b 100644
--- a/producer/option.go
+++ b/producer/option.go
@@ -122,8 +122,8 @@ func WithCreateTopicKey(topic string) Option {
 	}
 }
 
-// WithNsResovler set nameserver resolver to fetch nameserver addr
-func WithNsResovler(resolver primitive.NsResolver) Option {
+// WithNsResolver set nameserver resolver to fetch nameserver addr
+func WithNsResolver(resolver primitive.NsResolver) Option {
 	return func(options *producerOptions) {
 		options.Resolver = resolver
 	}


Test patch:
diff --git a/consumer/offset_store_test.go b/consumer/offset_store_test.go
index 27b98d9..cfa0eaa 100644
--- a/consumer/offset_store_test.go
+++ b/consumer/offset_store_test.go
@@ -21,12 +21,11 @@ import (
 	"path/filepath"
 	"testing"
 
-	"github.com/golang/mock/gomock"
-	. "github.com/smartystreets/goconvey/convey"
-
 	"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"
+	"github.com/golang/mock/gomock"
+	. "github.com/smartystreets/goconvey/convey"
 )
 
 func TestNewLocalFileOffsetStore(t *testing.T) {
@@ -136,7 +135,7 @@ func TestLocalFileOffsetStore(t *testing.T) {
 			offset = localStore.read(mq, _ReadFromStore)
 			So(offset, ShouldEqual, 1)
 
-			delete(localStore.(*localFileOffsetStore).OffsetTable, MessageQueueKey(*mq))
+			localStore.(*localFileOffsetStore).OffsetTable.Delete(MessageQueueKey(*mq))
 			offset = localStore.read(mq, _ReadMemoryThenStore)
 			So(offset, ShouldEqual, 1)
 		})
diff --git a/consumer/push_consumer_test.go b/consumer/push_consumer_test.go
index 7cb5fca..4789e9b 100644
--- a/consumer/push_consumer_test.go
+++ b/consumer/push_consumer_test.go
@@ -36,7 +36,7 @@ func TestStart(t *testing.T) {
 	Convey("test Start method", t, func() {
 		c, _ := NewPushConsumer(
 			WithGroupName("testGroup"),
-			WithNsResovler(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
+			WithNsResolver(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
 			WithConsumerModel(BroadCasting),
 		)
 
diff --git a/producer/producer_test.go b/producer/producer_test.go
index 508acf8..d1b88c3 100644
--- a/producer/producer_test.go
+++ b/producer/producer_test.go
@@ -35,7 +35,7 @@ const (
 
 func TestShutdown(t *testing.T) {
 	p, _ := NewDefaultProducer(
-		WithNsResovler(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
+		WithNsResolver(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
 		WithRetry(2),
 		WithQueueSelector(NewManualQueueSelector()),
 	)
@@ -98,7 +98,7 @@ func mockB4Send(p *defaultProducer) {
 
 func TestSync(t *testing.T) {
 	p, _ := NewDefaultProducer(
-		WithNsResovler(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
+		WithNsResolver(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
 		WithRetry(2),
 		WithQueueSelector(NewManualQueueSelector()),
 	)
@@ -149,7 +149,7 @@ func TestSync(t *testing.T) {
 
 func TestASync(t *testing.T) {
 	p, _ := NewDefaultProducer(
-		WithNsResovler(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
+		WithNsResolver(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
 		WithRetry(2),
 		WithQueueSelector(NewManualQueueSelector()),
 	)
@@ -211,7 +211,7 @@ func TestASync(t *testing.T) {
 
 func TestOneway(t *testing.T) {
 	p, _ := NewDefaultProducer(
-		WithNsResovler(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
+		WithNsResolver(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
 		WithRetry(2),
 		WithQueueSelector(NewManualQueueSelector()),
 	)