Skip to content

Commit bd8861c

Browse files
cytnjumuyun.cyt
and
muyun.cyt
authored
fix bug:seek offset won't work due to wrong map key type (#1184)
Co-authored-by: muyun.cyt <[email protected]>
1 parent abae4a0 commit bd8861c

File tree

1 file changed

+9
-2
lines changed

1 file changed

+9
-2
lines changed

consumer/pull_consumer.go

Lines changed: 9 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -208,12 +208,17 @@ func (pc *defaultPullConsumer) nextPullOffset(mq *primitive.MessageQueue, origin
208208
if pc.SubType != Assign {
209209
return originOffset
210210
}
211-
value, exist := pc.mq2seekOffset.LoadAndDelete(mq)
211+
value, exist := pc.mq2seekOffset.LoadAndDelete(*mq)
212212
if !exist {
213213
return originOffset
214214
} else {
215215
nextOffset := value.(int64)
216216
_ = pc.updateOffset(mq, nextOffset)
217+
rlog.Info("pull consumer assign new offset", map[string]interface{}{
218+
"group": pc.GroupName,
219+
"mq": mq,
220+
"offset": nextOffset,
221+
})
217222
return nextOffset
218223
}
219224
}
@@ -711,7 +716,7 @@ func (pc *defaultPullConsumer) ResetOffset(topic string, table map[primitive.Mes
711716
}
712717

713718
func (pc *defaultPullConsumer) SeekOffset(mq *primitive.MessageQueue, offset int64) {
714-
pc.mq2seekOffset.Store(mq, offset)
719+
pc.mq2seekOffset.Store(*mq, offset)
715720
rlog.Info("pull consumer seek offset", map[string]interface{}{
716721
"mq": mq,
717722
"offset": offset,
@@ -881,6 +886,8 @@ func (pc *defaultPullConsumer) pullMessage(request *PullRequest) {
881886
pullRequest.SysFlag = clearCommitOffsetFlag(pullRequest.SysFlag)
882887
}
883888

889+
rlog.Debug(fmt.Sprintf("defaultPullConsumer pull message from broker: %s, request: %+v", brokerResult.BrokerAddr, pullRequest), nil)
890+
884891
result, err := pc.client.PullMessage(context.Background(), brokerResult.BrokerAddr, pullRequest)
885892
if err != nil {
886893
rlog.Warning("defaultPullConsumer pull message from broker error", map[string]interface{}{

0 commit comments

Comments
 (0)