Merge branch 'main' into kip-580-exponential-backoff-func

Author: wanwenli

client.go

@@ -113,6 +113,9 @@ type Client interface {
 	// LeastLoadedBroker retrieves broker that has the least responses pending
 	LeastLoadedBroker() *Broker
 
+	// check if partition is readable
+	PartitionNotReadable(topic string, partition int32) bool
+
 	// Close shuts down all broker connections managed by this client. It is required
 	// to call this function before a client object passes out of scope, as it will
 	// otherwise leak memory. You must close any Producers or Consumers using a client
@@ -1283,3 +1286,14 @@ type nopCloserClient struct {
 func (ncc *nopCloserClient) Close() error {
 	return nil
 }
+
+func (client *client) PartitionNotReadable(topic string, partition int32) bool {
+	client.lock.RLock()
+	defer client.lock.RUnlock()
+
+	pm := client.metadata[topic][partition]
+	if pm == nil {
+		return true
+	}
+	return pm.Leader == -1
+}

consumer_group.go

@@ -861,18 +861,32 @@ func newConsumerGroupSession(ctx context.Context, parent *consumerGroup, claims
 		return nil, err
 	}
 
-	// start consuming
+	// start consuming each topic partition in its own goroutine
 	for topic, partitions := range claims {
 		for _, partition := range partitions {
-			sess.waitGroup.Add(1)
-
+			sess.waitGroup.Add(1) // increment wait group before spawning goroutine
 			go func(topic string, partition int32) {
 				defer sess.waitGroup.Done()
-
-				// cancel the as session as soon as the first
-				// goroutine exits
+				// cancel the group session as soon as any of the consume calls return
 				defer sess.cancel()
 
+				// if partition not currently readable, wait for it to become readable
+				if sess.parent.client.PartitionNotReadable(topic, partition) {
+					timer := time.NewTimer(5 * time.Second)
+					defer timer.Stop()
+
+					for sess.parent.client.PartitionNotReadable(topic, partition) {
+						select {
+						case <-ctx.Done():
+							return
+						case <-parent.closed:
+							return
+						case <-timer.C:
+							timer.Reset(5 * time.Second)
+						}
+					}
+				}
+
 				// consume a single topic/partition, blocking
 				sess.consume(topic, partition)
 			}(topic, partition)