kgo: fix race condition in close

twmb · May 15, 2023 · ee3d7c1 · ee3d7c1
1 parent 88d5c46
commit ee3d7c1
Showing 1 changed file with 2 additions and 2 deletions.
diff --git a/pkg/kgo/client.go b/pkg/kgo/client.go
@@ -948,7 +948,7 @@ func (cl *Client) Close() {
 
  sessCloseCtx, sessCloseCancel := context.WithTimeout(cl.ctx, time.Second)
  var wg sync.WaitGroup
- for _, sns := range cl.sinksAndSources {
+ cl.allSinksAndSources(func(sns sinkAndSource) {
  if sns.source.session.id != 0 {
  sns := sns
  wg.Add(1)
@@ -957,7 +957,7 @@ func (cl *Client) Close() {
  sns.source.killSessionOnClose(sessCloseCtx)
  }()
  }
- }
+ })
  wg.Wait()
  sessCloseCancel()