#133 fix race.
Some checks failed
autoapprove / autoapprove (pull_request) Has been cancelled
automerge / automerge (pull_request) Has been cancelled
dependabot-automerge / automerge (pull_request) Has been cancelled
codeql / analyze (go) (pull_request) Has been cancelled
prbuild / test (pull_request) Has been cancelled
prbuild / lint (pull_request) Has been cancelled
Some checks failed
autoapprove / autoapprove (pull_request) Has been cancelled
automerge / automerge (pull_request) Has been cancelled
dependabot-automerge / automerge (pull_request) Has been cancelled
codeql / analyze (go) (pull_request) Has been cancelled
prbuild / test (pull_request) Has been cancelled
prbuild / lint (pull_request) Has been cancelled
This commit is contained in:
parent
ceae272ca8
commit
26daa968b8
@ -105,11 +105,13 @@ func (s *subscriber) poll(ctx context.Context) {
|
||||
continue
|
||||
}
|
||||
|
||||
s.Lock()
|
||||
for tp := range s.consumers {
|
||||
if v, ok := lmap[tp.p]; ok {
|
||||
s.kopts.Meter.Counter(semconv.BrokerGroupLag, "topic", s.topic, "group", s.opts.Group, "partition", strconv.Itoa(int(tp.p)), "lag", strconv.Itoa(int(v.Lag)))
|
||||
}
|
||||
}
|
||||
s.Unlock()
|
||||
|
||||
}
|
||||
}
|
||||
|
Loading…
x
Reference in New Issue
Block a user