kgo: fix bug

twmb · twmb · commit cd65d77d05f7 · 2024-02-06T17:41:33.000-07:00
Previously, if CommitOffsetsSync was called during a rebalance, and the context being used is canceled while the rebalance was occurring, then the client would deadlock. Internally, committing is blocked if a rebalance is actively happening. There's some complex logic to have, effectively, a cancelable lock if the user wants to not wait for a rebalance to complete while trying to commit offsets. There was a bug, and since the fix is one line, it's easier to see than explain. Closes #668.
diff --git a/pkg/kgo/consumer_group.go b/pkg/kgo/consumer_group.go
@@ -2584,6 +2584,7 @@ func (g *groupConsumer) commitOffsetsSync(
 
 	if err := g.waitJoinSyncMu(ctx); err != nil {
 		onDone(g.cl, kmsg.NewPtrOffsetCommitRequest(), kmsg.NewPtrOffsetCommitResponse(), err)
+		close(done)
 		return
 	}
 	unblockCommits := func(cl *Client, req *kmsg.OffsetCommitRequest, resp *kmsg.OffsetCommitResponse, err error) {

Original file line number	Diff line number	Diff line change
`@@ -2584,6 +2584,7 @@ func (g *groupConsumer) commitOffsetsSync(`
`2584`	`2584`
`2585`	`2585`	`if err := g.waitJoinSyncMu(ctx); err != nil {`
`2586`	`2586`	`onDone(g.cl, kmsg.NewPtrOffsetCommitRequest(), kmsg.NewPtrOffsetCommitResponse(), err)`
	`2587`	`+ close(done)`
`2587`	`2588`	`return`
`2588`	`2589`	`}`
`2589`	`2590`	`unblockCommits := func(cl Client, req kmsg.OffsetCommitRequest, resp *kmsg.OffsetCommitResponse, err error) {`