From b538ef82b51838224bb7f2eb036517ed0cfab3aa Mon Sep 17 00:00:00 2001
From: pugnack <pugnack@gmail.com>
Date: Fri, 23 May 2025 01:45:39 +0500
Subject: [PATCH] [v4] hide access to internal mutex (#185)

* changed embedded mutex to private field

* update ci

* fix tests
---
 kgo.go        | 22 +++++++++++-----------
 subscriber.go | 12 ++++++------
 2 files changed, 17 insertions(+), 17 deletions(-)

diff --git a/kgo.go b/kgo.go
index c87ea02..6af9d39 100644
--- a/kgo.go
+++ b/kgo.go
@@ -74,7 +74,7 @@ type Broker struct {
 
 	opts broker.Options
 
-	sync.RWMutex
+	mu   sync.RWMutex
 	init bool
 }
 
@@ -141,9 +141,9 @@ func (b *Broker) newCodec(ct string) (codec.Codec, error) {
 	if idx := strings.IndexRune(ct, ';'); idx >= 0 {
 		ct = ct[:idx]
 	}
-	b.RLock()
+	b.mu.RLock()
 	c, ok := b.opts.Codecs[ct]
-	b.RUnlock()
+	b.mu.RUnlock()
 	if ok {
 		return c, nil
 	}
@@ -238,10 +238,10 @@ func (k *Broker) Connect(ctx context.Context) error {
 		return err
 	}
 
-	k.Lock()
+	k.mu.Lock()
 	k.c = c
 	k.connected.Store(1)
-	k.Unlock()
+	k.mu.Unlock()
 
 	return nil
 }
@@ -259,8 +259,8 @@ func (k *Broker) Disconnect(ctx context.Context) error {
 	ctx, span = k.opts.Tracer.Start(ctx, "Disconnect")
 	defer span.Finish()
 
-	k.Lock()
-	defer k.Unlock()
+	k.mu.Lock()
+	defer k.mu.Unlock()
 	select {
 	case <-nctx.Done():
 		return nctx.Err()
@@ -284,8 +284,8 @@ func (k *Broker) Disconnect(ctx context.Context) error {
 }
 
 func (k *Broker) Init(opts ...broker.Option) error {
-	k.Lock()
-	defer k.Unlock()
+	k.mu.Lock()
+	defer k.mu.Unlock()
 
 	if len(opts) == 0 && k.init {
 		return nil
@@ -538,9 +538,9 @@ func (b *Broker) fnSubscribe(ctx context.Context, topic string, handler interfac
 
 	go sub.poll(ctx)
 
-	b.Lock()
+	b.mu.Lock()
 	b.subs = append(b.subs, sub)
-	b.Unlock()
+	b.mu.Unlock()
 
 	return sub, nil
 }
diff --git a/subscriber.go b/subscriber.go
index e57177e..4aeb6b1 100644
--- a/subscriber.go
+++ b/subscriber.go
@@ -49,8 +49,8 @@ type Subscriber struct {
 	kopts       broker.Options
 	opts        broker.SubscribeOptions
 
-	connected *atomic.Uint32
-	sync.RWMutex
+	connected    *atomic.Uint32
+	mu           sync.RWMutex
 	closed       bool
 	fatalOnError bool
 }
@@ -118,11 +118,11 @@ func (s *Subscriber) poll(ctx context.Context) {
 					continue
 				}
 
-				s.Lock()
+				s.mu.Lock()
 				for p, l := range lmap {
 					s.kopts.Meter.Counter(semconv.BrokerGroupLag, "topic", s.topic, "group", s.opts.Group, "partition", strconv.Itoa(int(p))).Set(uint64(l.Lag))
 				}
-				s.Unlock()
+				s.mu.Unlock()
 
 			}
 		}
@@ -230,9 +230,9 @@ func (s *Subscriber) assigned(_ context.Context, c *kgo.Client, assigned map[str
 				opts:        s.opts,
 				connected:   s.connected,
 			}
-			s.Lock()
+			s.mu.Lock()
 			s.consumers[tp{topic, partition}] = pc
-			s.Unlock()
+			s.mu.Unlock()
 			go pc.consume()
 		}
 	}