export lag for all partition, not only owned

Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>

kgo.go

@@ -98,8 +98,10 @@ func (k *Broker) connect(ctx context.Context, opts ...kgo.Opt) (*kgo.Client, err
 	select {
 	case <-ctx.Done():
 		if ctx.Err() != nil {
+			if sp != nil {
 				sp.SetStatus(tracer.SpanStatusError, ctx.Err().Error())
 			}
+		}
 		return nil, ctx.Err()
 	default:
 		c, err = kgo.NewClient(opts...)
@@ -107,7 +109,9 @@ func (k *Broker) connect(ctx context.Context, opts ...kgo.Opt) (*kgo.Client, err
 			err = c.Ping(ctx) // check connectivity to cluster
 		}
 		if err != nil {
+			if sp != nil {
 				sp.SetStatus(tracer.SpanStatusError, err.Error())
+			}
 			return nil, err
 		}
 	}

subscriber.go

@@ -106,10 +106,8 @@ func (s *subscriber) poll(ctx context.Context) {
 				}
 
 				s.Lock()
-				for tp := range s.consumers {
+				for p, l := range lmap {
-					if v, ok := lmap[tp.p]; ok {
+					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(tp.p)), "lag", strconv.Itoa(int(v.Lag)))
-					}
 				}
 				s.Unlock()