unistack-org/micro-broker-kgo
4
0
Fork 2
Code Issues 3 Pull Requests 1 Actions 8 Packages Projects Releases 45 Wiki Activity

#133 fix race. #134

Merged
vtolstov merged 2 commits from kgorbunov/micro-broker-kgo:data-racev3 into v3 2024-04-19 19:26:07 +03:00
Conversation 0 Commits 2 Files Changed 1 +4
1 changed files with 4 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
subscriber.go
View File

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