export lag for all partition, not only owned

Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2024-05-02 23:01:04 +03:00 · 2024-05-02 23:01:04 +03:00 · 7329bc23bc
commit 7329bc23bc
parent c240631cdb
1 changed files with 2 additions and 4 deletions
--- a/subscriber.go
+++ b/subscriber.go
@ -106,10 +106,8 @@ func (s *subscriber) poll(ctx context.Context) {
 				}

 				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)))
-					}
+				for p, l := range lmap {
+					s.kopts.Meter.Counter(semconv.BrokerGroupLag, "topic", s.topic, "group", s.opts.Group, "partition", strconv.Itoa(int(p)), "lag", strconv.Itoa(int(l.Lag)))
 				}
 				s.Unlock()