fix deadlock when polling multiple times from Kafka

KiKoS0 · KiKoS0 · commit 94fd8622c6d6 · 2025-12-13T17:51:57.000+01:00
diff --git a/pkg/kgo/consumer.go b/pkg/kgo/consumer.go
@@ -239,8 +239,10 @@ func (c *consumer) waitAndAddPoller() {
 	}
 	c.pollWaitMu.Lock()
 	defer c.pollWaitMu.Unlock()
-	for c.pollWaitState>>32 != 0 {
-		c.pollWaitC.Wait()
+	if c.pollWaitState&math.MaxUint32 == 0 {
+		for c.pollWaitState>>32 != 0 {
+			c.pollWaitC.Wait()
+		}
 	}
 	// Rebalance always takes priority, but if there are no active
 	// rebalances, our poll blocks rebalances.

Original file line number	Diff line number	Diff line change
`@@ -239,8 +239,10 @@ func (c *consumer) waitAndAddPoller() {`
`239`	`239`	`}`
`240`	`240`	`c.pollWaitMu.Lock()`
`241`	`241`	`defer c.pollWaitMu.Unlock()`
`242`		`- for c.pollWaitState>>32 != 0 {`
`243`		`- c.pollWaitC.Wait()`
	`242`	`+ if c.pollWaitState&math.MaxUint32 == 0 {`
	`243`	`+ for c.pollWaitState>>32 != 0 {`
	`244`	`+ c.pollWaitC.Wait()`
	`245`	`+ }`
`244`	`246`	`}`
`245`	`247`	`// Rebalance always takes priority, but if there are no active`
`246`	`248`	`// rebalances, our poll blocks rebalances.`