diff --git a/subscriber.go b/subscriber.go index 44560d7..80d3587 100644 --- a/subscriber.go +++ b/subscriber.go @@ -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() } }