ConsumerGroup.Close can infinitely lock

IBM · Nov 9, 2023 · d610d0c · d610d0c
1 parent 01b853a
commit d610d0c
Showing 1 changed file with 6 additions and 2 deletions.
diff --git a/consumer_group.go b/consumer_group.go
@@ -213,8 +213,12 @@ func (c *consumerGroup) Consume(ctx context.Context, topics []string, handler Co
 		return err
 	}
 
-	// Wait for session exit signal
-	<-sess.ctx.Done()
+	// Wait for session exit signal or Close() call
+	select {
+	case <-c.closed:
+	case <-sess.ctx.Done():
+	default:
+	}
 
 	// Gracefully release session claims
 	return sess.release(true)