Skip to content

Commit

Permalink
kgo: further fix for cd65d77
Browse files Browse the repository at this point in the history
The prior commit was insufficient -- we left a dangling lock, and we
had that same dangling lock in the cancelable commit offsets.
  • Loading branch information
twmb committed Feb 7, 2024
1 parent 41f0269 commit 99d6dfb
Showing 1 changed file with 6 additions and 9 deletions.
15 changes: 6 additions & 9 deletions pkg/kgo/consumer_group.go
Original file line number Diff line number Diff line change
Expand Up @@ -2580,13 +2580,13 @@ func (g *groupConsumer) commitOffsetsSync(
onDone = func(*Client, *kmsg.OffsetCommitRequest, *kmsg.OffsetCommitResponse, error) {}
}

g.syncCommitMu.Lock() // block all other concurrent commits until our OnDone is done.

if err := g.waitJoinSyncMu(ctx); err != nil {
onDone(g.cl, kmsg.NewPtrOffsetCommitRequest(), kmsg.NewPtrOffsetCommitResponse(), err)
close(done)
return
}

g.syncCommitMu.Lock() // block all other concurrent commits until our OnDone is done.
unblockCommits := func(cl *Client, req *kmsg.OffsetCommitRequest, resp *kmsg.OffsetCommitResponse, err error) {
g.noCommitDuringJoinAndSync.RUnlock()
defer close(done)
Expand Down Expand Up @@ -2663,19 +2663,16 @@ func (cl *Client) CommitOffsets(
return
}

g.syncCommitMu.RLock() // block sync commit, but allow other concurrent Commit to cancel us
unblockSyncCommit := func(cl *Client, req *kmsg.OffsetCommitRequest, resp *kmsg.OffsetCommitResponse, err error) {
defer g.syncCommitMu.RUnlock()
onDone(cl, req, resp, err)
}

if err := g.waitJoinSyncMu(ctx); err != nil {
onDone(g.cl, kmsg.NewPtrOffsetCommitRequest(), kmsg.NewPtrOffsetCommitResponse(), err)
return
}

g.syncCommitMu.RLock() // block sync commit, but allow other concurrent Commit to cancel us
unblockJoinSync := func(cl *Client, req *kmsg.OffsetCommitRequest, resp *kmsg.OffsetCommitResponse, err error) {
g.noCommitDuringJoinAndSync.RUnlock()
unblockSyncCommit(cl, req, resp, err)
defer g.syncCommitMu.RUnlock()
onDone(cl, req, resp, err)
}

g.mu.Lock()
Expand Down

0 comments on commit 99d6dfb

Please sign in to comment.