Compare commits

...

2 Commits

Author SHA1 Message Date
f73d9d7347 fixup group lag metric
All checks were successful
test / test (push) Successful in 3m49s
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2025-04-23 10:51:39 +03:00
53f72f1246 fixup panic
All checks were successful
test / test (push) Successful in 1m57s
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2025-03-12 20:34:30 +03:00

View File

@@ -119,7 +119,7 @@ func (s *Subscriber) poll(ctx context.Context) {
s.Lock()
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.kopts.Meter.Counter(semconv.BrokerGroupLag, "topic", s.topic, "group", s.opts.Group, "partition", strconv.Itoa(int(p))).Set(uint64(l.Lag))
}
s.Unlock()
@@ -160,7 +160,10 @@ func (s *Subscriber) killConsumers(ctx context.Context, lost map[string][]int32)
for topic, partitions := range lost {
for _, partition := range partitions {
tps := tp{topic, partition}
pc := s.consumers[tps]
pc, ok := s.consumers[tps]
if !ok {
continue
}
delete(s.consumers, tps)
close(pc.quit)
if s.kopts.Logger.V(logger.DebugLevel) {