export lag for all partition, not only owned

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

kgo.go

@@ -98,7 +98,9 @@ func (k *Broker) connect(ctx context.Context, opts ...kgo.Opt) (*kgo.Client, err
 	select {
 	case <-ctx.Done():
 		if ctx.Err() != nil {
-			sp.SetStatus(tracer.SpanStatusError, ctx.Err().Error())
+			if sp != nil {
+				sp.SetStatus(tracer.SpanStatusError, ctx.Err().Error())
+			}
 		}
 		return nil, ctx.Err()
 	default:
@@ -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 {
-			sp.SetStatus(tracer.SpanStatusError, err.Error())
+			if sp != nil {
+				sp.SetStatus(tracer.SpanStatusError, err.Error())
+			}
 			return nil, err
 		}
 	}

subscriber.go

@@ -105,11 +105,11 @@ func (s *subscriber) poll(ctx context.Context) {
 					continue
 				}
 
-				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.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.Unlock()
 
 			}
 		}
@@ -186,7 +186,9 @@ func (s *subscriber) assigned(_ context.Context, c *kgo.Client, assigned map[str
 				kopts:   s.kopts,
 				opts:    s.opts,
 			}
+			s.Lock()
 			s.consumers[tp{topic, partition}] = pc
+			s.Unlock()
 			go pc.consume()
 		}
 	}