Repository: apache/rocketmq-client-go
Pull request #841 (2022-06-14 03:43:56)

[ISSUE #823]fix: panic problem caused by json-iterator in go1.18
## What is the purpose of the change

fixing client panic problem caused by low version of `json-iterator`

## Brief changelog

update go version and `json-iterator` package version to latest.

## Verifying this change

`json-iterator` repo has publish `v1.1.12` to fix panic problem in go go1.18
<img width="1782" alt="image" src="https://user-images.githubusercontent.com/21994267/173488899-39b62051-8586-4962-ba49-14daa6bd27fe.png">


yusank: #823

Patch:
diff --git a/consumer/consumer.go b/consumer/consumer.go
index ed843cf..6e53eaf 100644
--- a/consumer/consumer.go
+++ b/consumer/consumer.go
@@ -28,6 +28,7 @@ import (
 	"time"
 
 	"github.com/apache/rocketmq-client-go/v2/errors"
+
 	jsoniter "github.com/json-iterator/go"
 	"github.com/tidwall/gjson"
 
@@ -699,8 +700,9 @@ func (dc *defaultConsumer) updateProcessQueueTable(topic string, mqs []*primitiv
 				continue
 			}
 			dc.storage.remove(&mq)
-			nextOffset := dc.computePullFromWhere(&mq)
-			if nextOffset >= 0 {
+			nextOffset, err := dc.computePullFromWhereWithException(&mq)
+
+			if nextOffset >= 0 && err == nil {
 				_, exist := dc.processQueueTable.Load(mq)
 				if exist {
 					rlog.Debug("do defaultConsumer, mq already exist", map[string]interface{}{
@@ -741,12 +743,23 @@ func (dc *defaultConsumer) removeUnnecessaryMessageQueue(mq *primitive.MessageQu
 	return true
 }
 
+// Deprecated: Use computePullFromWhereWithException instead.
 func (dc *defaultConsumer) computePullFromWhere(mq *primitive.MessageQueue) int64 {
+	result, _ := dc.computePullFromWhereWithException(mq)
+	return result
+}
+
+func (dc *defaultConsumer) computePullFromWhereWithException(mq *primitive.MessageQueue) (int64, error) {
 	if dc.cType == _PullConsume {
-		return 0
+		return 0, nil
 	}
-	var result = int64(-1)
-	lastOffset := dc.storage.read(mq, _ReadFromStore)
+	result := int64(-1)
+	lastOffset, err := dc.storage.readWithException(mq, _ReadFromStore)
+	if err != nil {
+		// 这里 lastOffset = -1
+		return lastOffset, err
+	}
+
 	if lastOffset >= 0 {
 		result = lastOffset
 	} else {
@@ -803,7 +816,7 @@ func (dc *defaultConsumer) computePullFromWhere(mq *primitive.MessageQueue) int6
 		default:
 		}
 	}
-	return result
+	return result, nil
 }
 
 func (dc *defaultConsumer) pullInner(ctx context.Context, queue *primitive.MessageQueue, data *internal.SubscriptionData,
@@ -950,7 +963,8 @@ func (dc *defaultConsumer) queryMaxOffset(mq *primitive.MessageQueue) (int64, er
 }
 
 func (dc *defaultConsumer) queryOffset(mq *primitive.MessageQueue) int64 {
-	return dc.storage.read(mq, _ReadMemoryThenStore)
+	result, _ := dc.storage.readWithException(mq, _ReadMemoryThenStore)
+	return result
 }
 
 // SearchOffsetByTimestamp with specific queueId and topic
diff --git a/consumer/mock_offset_store.go b/consumer/mock_offset_store.go
index bac1cdb..145ec3d 100644
--- a/consumer/mock_offset_store.go
+++ b/consumer/mock_offset_store.go
@@ -28,67 +28,90 @@ import (
 	gomock "github.com/golang/mock/gomock"
 )
 
-// MockOffsetStore is a mock of OffsetStore interface
+// MockOffsetStore is a mock of OffsetStore interface.
 type MockOffsetStore struct {
 	ctrl     *gomock.Controller
 	recorder *MockOffsetStoreMockRecorder
 }
 
-// MockOffsetStoreMockRecorder is the mock recorder for MockOffsetStore
+// MockOffsetStoreMockRecorder is the mock recorder for MockOffsetStore.
 type MockOffsetStoreMockRecorder struct {
 	mock *MockOffsetStore
 }
 
-// NewMockOffsetStore creates a new mock instance
+// NewMockOffsetStore creates a new mock instance.
 func NewMockOffsetStore(ctrl *gomock.Controller) *MockOffsetStore {
 	mock := &MockOffsetStore{ctrl: ctrl}
 	mock.recorder = &MockOffsetStoreMockRecorder{mock}
 	return mock
 }
 
-// EXPECT returns an object that allows the caller to indicate expected use
+// EXPECT returns an object that allows the caller to indicate expected use.
 func (m *MockOffsetStore) EXPECT() *MockOffsetStoreMockRecorder {
 	return m.recorder
 }
 
-// persist mocks base method
+// persist mocks base method.
 func (m *MockOffsetStore) persist(mqs []*primitive.MessageQueue) {
+	m.ctrl.T.Helper()
 	m.ctrl.Call(m, "persist", mqs)
 }
 
-// persist indicates an expected call of persist
+// persist indicates an expected call of persist.
 func (mr *MockOffsetStoreMockRecorder) persist(mqs interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "persist", reflect.TypeOf((*MockOffsetStore)(nil).persist), mqs)
 }
 
-// remove mocks base method
-func (m *MockOffsetStore) remove(mq *primitive.MessageQueue) {
-	m.ctrl.Call(m, "remove", mq)
-}
-
-// remove indicates an expected call of remove
-func (mr *MockOffsetStoreMockRecorder) remove(mq interface{}) *gomock.Call {
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "remove", reflect.TypeOf((*MockOffsetStore)(nil).remove), mq)
-}
-
-// read mocks base method
+// read mocks base method.
 func (m *MockOffsetStore) read(mq *primitive.MessageQueue, t readType) int64 {
+	m.ctrl.T.Helper()
 	ret := m.ctrl.Call(m, "read", mq, t)
 	ret0, _ := ret[0].(int64)
 	return ret0
 }
 
-// read indicates an expected call of read
+// read indicates an expected call of read.
 func (mr *MockOffsetStoreMockRecorder) read(mq, t interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "read", reflect.TypeOf((*MockOffsetStore)(nil).read), mq, t)
 }
 
-// update mocks base method
+// readWithException mocks base method.
+func (m *MockOffsetStore) readWithException(mq *primitive.MessageQueue, t readType) (int64, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "readWithException", mq, t)
+	ret0, _ := ret[0].(int64)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// readWithException indicates an expected call of readWithException.
+func (mr *MockOffsetStoreMockRecorder) readWithException(mq, t interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "readWithException", reflect.TypeOf((*MockOffsetStore)(nil).readWithException), mq, t)
+}
+
+// remove mocks base method.
+func (m *MockOffsetStore) remove(mq *primitive.MessageQueue) {
+	m.ctrl.T.Helper()
+	m.ctrl.Call(m, "remove", mq)
+}
+
+// remove indicates an expected call of remove.
+func (mr *MockOffsetStoreMockRecorder) remove(mq interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "remove", reflect.TypeOf((*MockOffsetStore)(nil).remove), mq)
+}
+
+// update mocks base method.
 func (m *MockOffsetStore) update(mq *primitive.MessageQueue, offset int64, increaseOnly bool) {
+	m.ctrl.T.Helper()
 	m.ctrl.Call(m, "update", mq, offset, increaseOnly)
 }
 
-// update indicates an expected call of update
+// update indicates an expected call of update.
 func (mr *MockOffsetStoreMockRecorder) update(mq, offset, increaseOnly interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "update", reflect.TypeOf((*MockOffsetStore)(nil).update), mq, offset, increaseOnly)
 }
diff --git a/consumer/offset_store.go b/consumer/offset_store.go
index 5ecfd14..86ecd18 100644
--- a/consumer/offset_store.go
+++ b/consumer/offset_store.go
@@ -57,6 +57,7 @@ type OffsetStore interface {
 	persist(mqs []*primitive.MessageQueue)
 	remove(mq *primitive.MessageQueue)
 	read(mq *primitive.MessageQueue, t readType) int64
+	readWithException(mq *primitive.MessageQueue, t readType) (int64, error)
 	update(mq *primitive.MessageQueue, offset int64, increaseOnly bool)
 }
 
@@ -156,21 +157,27 @@ func (local *localFileOffsetStore) load() {
 	}
 }
 
+// Deprecated: Use readWithException instead.
 func (local *localFileOffsetStore) read(mq *primitive.MessageQueue, t readType) int64 {
+	result, _ := local.readWithException(mq, t)
+	return result
+}
+
+func (local *localFileOffsetStore) readWithException(mq *primitive.MessageQueue, t readType) (int64, error) {
 	switch t {
 	case _ReadFromMemory, _ReadMemoryThenStore:
 		off := readFromMemory(local.OffsetTable, mq)
 		if off >= 0 || (off == -1 && t == _ReadFromMemory) {
-			return off
+			return off, nil
 		}
 		fallthrough
 	case _ReadFromStore:
 		local.load()
-		return readFromMemory(local.OffsetTable, mq)
+		return readFromMemory(local.OffsetTable, mq), nil
 	default:
 
 	}
-	return -1
+	return -1, nil
 }
 
 func (local *localFileOffsetStore) update(mq *primitive.MessageQueue, offset int64, increaseOnly bool) {
@@ -284,18 +291,24 @@ func (r *remoteBrokerOffsetStore) remove(mq *primitive.MessageQueue) {
 	})
 }
 
+// Deprecated: Use readWithException instead.
 func (r *remoteBrokerOffsetStore) read(mq *primitive.MessageQueue, t readType) int64 {
+	result, _ := r.readWithException(mq, t)
+	return result
+}
+
+func (r *remoteBrokerOffsetStore) readWithException(mq *primitive.MessageQueue, t readType) (int64, error) {
 	r.mutex.RLock()
 	switch t {
 	case _ReadFromMemory, _ReadMemoryThenStore:
 		off, exist := r.OffsetTable[*mq]
 		if exist {
 			r.mutex.RUnlock()
-			return off
+			return off, nil
 		}
 		if t == _ReadFromMemory {
 			r.mutex.RUnlock()
-			return -1
+			return -1, nil
 		}
 		fallthrough
 	case _ReadFromStore:
@@ -307,7 +320,7 @@ func (r *remoteBrokerOffsetStore) read(mq *primitive.MessageQueue, t readType) i
 				rlog.LogKeyUnderlayError: err,
 			})
 			r.mutex.RUnlock()
-			return -1
+			return -1, err
 		}
 		rlog.Warning("fetch offset of mq from broker success", map[string]interface{}{
 			rlog.LogKeyConsumerGroup: r.group,
@@ -316,11 +329,11 @@ func (r *remoteBrokerOffsetStore) read(mq *primitive.MessageQueue, t readType) i
 		})
 		r.mutex.RUnlock()
 		r.update(mq, off, true)
-		return off
+		return off, nil
 	default:
 	}
 
-	return -1
+	return -1, nil
 }
 
 func (r *remoteBrokerOffsetStore) update(mq *primitive.MessageQueue, offset int64, increaseOnly bool) {
diff --git a/consumer/pull_consumer.go b/consumer/pull_consumer.go
index ff6f7d7..4699601 100644
--- a/consumer/pull_consumer.go
+++ b/consumer/pull_consumer.go
@@ -20,10 +20,11 @@ package consumer
 import (
 	"context"
 	"fmt"
-	errors2 "github.com/apache/rocketmq-client-go/v2/errors"
 	"sync"
 	"sync/atomic"
 
+	errors2 "github.com/apache/rocketmq-client-go/v2/errors"
+
 	"github.com/pkg/errors"
 
 	"github.com/apache/rocketmq-client-go/v2/internal"
@@ -219,7 +220,8 @@ func (c *defaultPullConsumer) makeSureStateOK() error {
 }
 
 func (c *defaultPullConsumer) nextOffsetOf(queue *primitive.MessageQueue) int64 {
-	return c.computePullFromWhere(queue)
+	result, _ := c.computePullFromWhereWithException(queue)
+	return result
 }
 
 // PullFrom pull messages of queue from the offset to offset + numbers
diff --git a/consumer/push_consumer.go b/consumer/push_consumer.go
index 801f412..4ad5ee3 100644
--- a/consumer/push_consumer.go
+++ b/consumer/push_consumer.go
@@ -20,7 +20,6 @@ package consumer
 import (
 	"context"
 	"fmt"
-	errors2 "github.com/apache/rocketmq-client-go/v2/errors"
 	"math"
 	"runtime/pprof"
 	"strconv"
@@ -29,6 +28,8 @@ import (
 	"sync/atomic"
 	"time"
 
+	errors2 "github.com/apache/rocketmq-client-go/v2/errors"
+
 	"github.com/pkg/errors"
 
 	"github.com/apache/rocketmq-client-go/v2/internal"
@@ -375,7 +376,7 @@ func (pc *pushConsumer) GetConsumerRunningInfo(stack bool) *internal.ConsumerRun
 		mq := key.(primitive.MessageQueue)
 		pq := value.(*processQueue)
 		pInfo := pq.currentInfo()
-		pInfo.CommitOffset = pc.storage.read(&mq, _ReadMemoryThenStore)
+		pInfo.CommitOffset, _ = pc.storage.readWithException(&mq, _ReadMemoryThenStore)
 		info.MQTable[mq] = pInfo
 		return true
 	})
@@ -644,7 +645,15 @@ func (pc *pushConsumer) pullMessage(request *PullRequest) {
 		} else {
 			if pq.IsLock() {
 				if !request.lockedFirst {
-					offset := pc.computePullFromWhere(request.mq)
+					offset, err := pc.computePullFromWhereWithException(request.mq)
+					if err != nil {
+						rlog.Warning("computePullFromWhere from broker error", map[string]interface{}{
+							rlog.LogKeyUnderlayError: err.Error(),
+						})
+						sleepTime = _PullDelayTimeWhenError
+						goto NEXT
+					}
+
 					brokerBusy := offset < request.nextOffset
 					rlog.Info("the first time to pull message, so fix offset from broker, offset maybe changed", map[string]interface{}{
 						rlog.LogKeyPullRequest:      request.String(),
@@ -684,7 +693,7 @@ func (pc *pushConsumer) pullMessage(request *PullRequest) {
 		)
 
 		if pc.model == Clustering {
-			commitOffsetValue = pc.storage.read(request.mq, _ReadFromMemory)
+			commitOffsetValue, _ = pc.storage.readWithException(request.mq, _ReadFromMemory)
 			if commitOffsetValue > 0 {
 				commitOffsetEnable = true
 			}
diff --git a/go.mod b/go.mod
index 6780484..93174fc 100644
--- a/go.mod
+++ b/go.mod
@@ -3,9 +3,10 @@ module github.com/apache/rocketmq-client-go/v2
 go 1.13
 
 require (
+	github.com/BurntSushi/toml v1.1.0 // indirect
 	github.com/emirpasic/gods v1.12.0
 	github.com/golang/mock v1.3.1
-	github.com/json-iterator/go v1.1.9
+	github.com/json-iterator/go v1.1.12
 	github.com/patrickmn/go-cache v2.1.0+incompatible
 	github.com/pkg/errors v0.8.1
 	github.com/satori/go.uuid v1.2.0
@@ -15,8 +16,8 @@ require (
 	github.com/stretchr/testify v1.3.0
 	github.com/tidwall/gjson v1.13.0
 	go.uber.org/atomic v1.5.1
-	golang.org/dl v0.0.0-20220315170520-faa7218da89a // indirect
-	golang.org/x/sys v0.0.0-20220319134239-a9b59b0215f8 // indirect
 	gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c // indirect
+	gopkg.in/natefinch/lumberjack.v2 v2.0.0
+	gopkg.in/yaml.v2 v2.4.0 // indirect
 	stathat.com/c/consistent v1.0.0 // indirect
 )
diff --git a/go.sum b/go.sum
index 5b3e9a3..b27a55a 100644
--- a/go.sum
+++ b/go.sum
@@ -1,3 +1,5 @@
+github.com/BurntSushi/toml v1.1.0 h1:ksErzDEI1khOiGPgpwuI7x2ebx/uXQNw7xJpn9Eq1+I=
+github.com/BurntSushi/toml v1.1.0/go.mod h1:CxXYINrC8qIiEnFrOxCa7Jy5BFHlXnUU2pbicEuybxQ=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
@@ -8,8 +10,8 @@ github.com/golang/mock v1.3.1/go.mod h1:sBzyDLLjw3U8JLTeZvSv8jJB+tU5PVekmnlKIyFU
 github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
 github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1 h1:EGx4pi6eqNxGaHF6qqu48+N2wcFQ5qg5FXgOdqsJ5d8=
 github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1/go.mod h1:wJfORRmW1u3UXTncJ5qlYoELFm8eSnnEO6hX4iZ3EWY=
-github.com/json-iterator/go v1.1.9 h1:9yzud/Ht36ygwatGx56VwCZtlI/2AD15T1X2sjSuGns=
-github.com/json-iterator/go v1.1.9/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4=
+github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM=
+github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo=
 github.com/jtolds/gls v4.20.0+incompatible h1:xdiiI2gbIgH/gLH7ADydsJ1uDOEzR8yvV7C0MuV77Wo=
 github.com/jtolds/gls v4.20.0+incompatible/go.mod h1:QJZ7F/aHp+rZTRtaJ1ow/lLfFfVYBRgL+9YlvaHOwJU=
 github.com/konsorten/go-windows-terminal-sequences v1.0.1 h1:mweAR1A6xJ3oS2pRaGiHgQ4OO8tzTaLawm8vnODuwDk=
@@ -21,8 +23,8 @@ github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
 github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421 h1:ZqeYNhU3OHLH3mGKHDcjJRFFRrJa6eAM5H+CtDdOsPc=
 github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
-github.com/modern-go/reflect2 v0.0.0-20180701023420-4b7aa43c6742 h1:Esafd1046DLDQ0W1YjYsBW+p8U2u7vzgW2SQVmlNazg=
-github.com/modern-go/reflect2 v0.0.0-20180701023420-4b7aa43c6742/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0=
+github.com/modern-go/reflect2 v1.0.2 h1:xBagoLtFs94CBntxluKeaWgTMpvLxC4ur3nMaC9Gz0M=
+github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk=
 github.com/patrickmn/go-cache v2.1.0+incompatible h1:HRMgzkcYKYpi3C8ajMPV8OFXaaRUnok+kx1WdO15EQc=
 github.com/patrickmn/go-cache v2.1.0+incompatible/go.mod h1:3Qf8kWWT7OJRJbdiICTKqZju1ZixQ/KpMGzzAfe6+WQ=
 github.com/pkg/errors v0.8.1 h1:iURUrRGxPUNPdy5/HRSm+Yj6okJ6UtLINN0Q9M4+h3I=
@@ -52,8 +54,6 @@ github.com/tidwall/pretty v1.2.0 h1:RWIZEg2iJ8/g6fDDYzMpobmaoGh5OLl4AXtGUGPcqCs=
 github.com/tidwall/pretty v1.2.0/go.mod h1:ITEVvHYasfjBbM0u2Pg8T2nJnzm8xPwvNhhsoaGGjNU=
 go.uber.org/atomic v1.5.1 h1:rsqfU5vBkVknbhUGbAUwQKR2H4ItV8tjJ+6kJX4cxHM=
 go.uber.org/atomic v1.5.1/go.mod h1:sABNBOSYdrvTF6hTgEIbc7YasKWGhgEQZyfxyTvoXHQ=
-golang.org/dl v0.0.0-20220315170520-faa7218da89a h1:9XQmLUvVWgdvCsxkaOk6rHC3dvrxOeMU+e8N4KikV7o=
-golang.org/dl v0.0.0-20220315170520-faa7218da89a/go.mod h1:IUMfjQLJQd4UTqG1Z90tenwKoCX93Gn3MAQJMOSBsDQ=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/lint v0.0.0-20190930215403-16217165b5de h1:5hukYrvBGR8/eNkX5mdUezrA6JiaEZDtJb9Ei+1LlBs=
 golang.org/x/lint v0.0.0-20190930215403-16217165b5de/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc=
@@ -61,9 +61,8 @@ golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn
 golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a h1:1BGLXjeY4akVXGgbC9HugT3Jv3hCI0z56oJR5vAMgBU=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
-golang.org/x/sys v0.0.0-20220319134239-a9b59b0215f8 h1:OH54vjqzRWmbJ62fjuhxy7AxFFgoHN0/DPc/UrL8cAs=
-golang.org/x/sys v0.0.0-20220319134239-a9b59b0215f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/tools v0.0.0-20190311212946-11955173bddd/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
 golang.org/x/tools v0.0.0-20190328211700-ab21143f2384/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
@@ -71,9 +70,12 @@ golang.org/x/tools v0.0.0-20190425150028-36563e24a262/go.mod h1:RgjU9mgBXZiqYHBn
 golang.org/x/tools v0.0.0-20191029041327-9cc4af7d6b2c h1:IGkKhmfzcztjm6gYkykvu/NiS8kaqbCWAEWWAyf8J5U=
 golang.org/x/tools v0.0.0-20191029041327-9cc4af7d6b2c/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
 gopkg.in/natefinch/lumberjack.v2 v2.0.0 h1:1Lc07Kr7qY4U2YPouBjpCLxpiyxIVoxqXgkXLknAOE8=
 gopkg.in/natefinch/lumberjack.v2 v2.0.0/go.mod h1:l0ndWWf7gzL7RNwBG7wST/UCcT4T24xpD6X8LsfU/+k=
+gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY=
+gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ=
 stathat.com/c/consistent v1.0.0 h1:ezyc51EGcRPJUxfHGSgJjWzJdj3NiMU9pNfLNGiXV0c=
 stathat.com/c/consistent v1.0.0/go.mod h1:QkzMWzcbB+yQBL2AttO6sgsQS/JSTapcDISJalmCDS0=


Test patch:
diff --git a/consumer/consumer_test.go b/consumer/consumer_test.go
index 12ccd18..e441290 100644
--- a/consumer/consumer_test.go
+++ b/consumer/consumer_test.go
@@ -26,6 +26,7 @@ import (
 	. "github.com/smartystreets/goconvey/convey"
 	"github.com/stretchr/testify/assert"
 
+	"github.com/apache/rocketmq-client-go/v2/errors"
 	"github.com/apache/rocketmq-client-go/v2/internal"
 	"github.com/apache/rocketmq-client-go/v2/internal/remote"
 	"github.com/apache/rocketmq-client-go/v2/primitive"
@@ -116,13 +117,20 @@ func TestComputePullFromWhere(t *testing.T) {
 		rmqCli.SetNameSrv(namesrvCli)
 
 		Convey("get effective offset", func() {
-			offsetStore.EXPECT().read(gomock.Any(), gomock.Any()).Return(int64(10))
-			res := dc.computePullFromWhere(mq)
+			offsetStore.EXPECT().readWithException(gomock.Any(), gomock.Any()).Return(int64(10), nil)
+			res, _ := dc.computePullFromWhereWithException(mq)
 			assert.Equal(t, int64(10), res)
 		})
 
+		Convey("get offset error", func() {
+			offsetStore.EXPECT().readWithException(gomock.Any(), gomock.Any()).Return(int64(-1), errors.ErrRequestTimeout)
+
+			_, err := dc.computePullFromWhereWithException(mq)
+			assert.Equal(t, err, errors.ErrRequestTimeout)
+		})
+
 		Convey("ConsumeFromLastOffset for normal topic", func() {
-			offsetStore.EXPECT().read(gomock.Any(), gomock.Any()).Return(int64(-1))
+			offsetStore.EXPECT().readWithException(gomock.Any(), gomock.Any()).Return(int64(-1), nil)
 			dc.option.FromWhere = ConsumeFromLastOffset
 
 			broker := "a"
@@ -135,20 +143,20 @@ func TestComputePullFromWhere(t *testing.T) {
 					},
 				}, nil)
 
-			res := dc.computePullFromWhere(mq)
+			res, _ := dc.computePullFromWhereWithException(mq)
 			assert.Equal(t, int64(20), res)
 		})
 
 		Convey("ConsumeFromFirstOffset for normal topic", func() {
-			offsetStore.EXPECT().read(gomock.Any(), gomock.Any()).Return(int64(-1))
+			offsetStore.EXPECT().readWithException(gomock.Any(), gomock.Any()).Return(int64(-1), nil)
 			dc.option.FromWhere = ConsumeFromFirstOffset
 
-			res := dc.computePullFromWhere(mq)
+			res, _ := dc.computePullFromWhereWithException(mq)
 			assert.Equal(t, int64(0), res)
 		})
 
 		Convey("ConsumeFromTimestamp for normal topic", func() {
-			offsetStore.EXPECT().read(gomock.Any(), gomock.Any()).Return(int64(-1))
+			offsetStore.EXPECT().readWithException(gomock.Any(), gomock.Any()).Return(int64(-1), nil)
 			dc.option.FromWhere = ConsumeFromTimestamp
 
 			dc.option.ConsumeTimestamp = "20060102150405"
@@ -163,7 +171,7 @@ func TestComputePullFromWhere(t *testing.T) {
 					},
 				}, nil)
 
-			res := dc.computePullFromWhere(mq)
+			res, _ := dc.computePullFromWhereWithException(mq)
 			assert.Equal(t, int64(30), res)
 		})
 
diff --git a/consumer/offset_store_test.go b/consumer/offset_store_test.go
index cfa0eaa..d833f36 100644
--- a/consumer/offset_store_test.go
+++ b/consumer/offset_store_test.go
@@ -98,7 +98,7 @@ func TestLocalFileOffsetStore(t *testing.T) {
 				}
 				for _, value := range cases {
 					localStore.update(value.queue, value.setOffset, false)
-					offset := localStore.read(value.queue, _ReadFromMemory)
+					offset, _ := localStore.readWithException(value.queue, _ReadFromMemory)
 					So(offset, ShouldEqual, value.expectedOffset)
 				}
 			})
@@ -119,7 +119,7 @@ func TestLocalFileOffsetStore(t *testing.T) {
 				}
 				for _, value := range cases {
 					localStore.update(value.queue, value.setOffset, true)
-					offset := localStore.read(value.queue, _ReadFromMemory)
+					offset, _ := localStore.readWithException(value.queue, _ReadFromMemory)
 					So(offset, ShouldEqual, value.expectedOffset)
 				}
 			})
@@ -127,16 +127,16 @@ func TestLocalFileOffsetStore(t *testing.T) {
 
 		Convey("test persist", func() {
 			localStore.update(mq, 1, false)
-			offset := localStore.read(mq, _ReadFromMemory)
+			offset, _ := localStore.readWithException(mq, _ReadFromMemory)
 			So(offset, ShouldEqual, 1)
 
 			queues := []*primitive.MessageQueue{mq}
 			localStore.persist(queues)
-			offset = localStore.read(mq, _ReadFromStore)
+			offset, _ = localStore.readWithException(mq, _ReadFromStore)
 			So(offset, ShouldEqual, 1)
 
 			localStore.(*localFileOffsetStore).OffsetTable.Delete(MessageQueueKey(*mq))
-			offset = localStore.read(mq, _ReadMemoryThenStore)
+			offset, _ = localStore.readWithException(mq, _ReadMemoryThenStore)
 			So(offset, ShouldEqual, 1)
 		})
 	})
@@ -178,7 +178,7 @@ func TestRemoteBrokerOffsetStore(t *testing.T) {
 				}
 				for _, value := range cases {
 					remoteStore.update(value.queue, value.setOffset, false)
-					offset := remoteStore.read(value.queue, _ReadFromMemory)
+					offset, _ := remoteStore.readWithException(value.queue, _ReadFromMemory)
 					So(offset, ShouldEqual, value.expectedOffset)
 				}
 			})
@@ -199,7 +199,7 @@ func TestRemoteBrokerOffsetStore(t *testing.T) {
 				}
 				for _, value := range cases {
 					remoteStore.update(value.queue, value.setOffset, true)
-					offset := remoteStore.read(value.queue, _ReadFromMemory)
+					offset, _ := remoteStore.readWithException(value.queue, _ReadFromMemory)
 					So(offset, ShouldEqual, value.expectedOffset)
 				}
 			})
@@ -219,24 +219,24 @@ func TestRemoteBrokerOffsetStore(t *testing.T) {
 			rmqClient.EXPECT().InvokeSync(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(ret, nil).MaxTimes(2)
 
 			remoteStore.persist(queues)
-			offset := remoteStore.read(mq, _ReadFromStore)
+			offset, _ := remoteStore.readWithException(mq, _ReadFromStore)
 			So(offset, ShouldEqual, 1)
 
 			remoteStore.remove(mq)
-			offset = remoteStore.read(mq, _ReadFromMemory)
+			offset, _ = remoteStore.readWithException(mq, _ReadFromMemory)
 			So(offset, ShouldEqual, -1)
-			offset = remoteStore.read(mq, _ReadMemoryThenStore)
+			offset, _ = remoteStore.readWithException(mq, _ReadMemoryThenStore)
 			So(offset, ShouldEqual, 1)
 
 		})
 
 		Convey("test remove", func() {
 			remoteStore.update(mq, 1, false)
-			offset := remoteStore.read(mq, _ReadFromMemory)
+			offset, _ := remoteStore.readWithException(mq, _ReadFromMemory)
 			So(offset, ShouldEqual, 1)
 
 			remoteStore.remove(mq)
-			offset = remoteStore.read(mq, _ReadFromMemory)
+			offset, _ = remoteStore.readWithException(mq, _ReadFromMemory)
 			So(offset, ShouldEqual, -1)
 		})
 	})