fix(consumer): add recovery from no leader partitions (#3101)

When some topic partitions have no leader due to Kafka broker failures,
the Sarama consumer group should be able to continue consuming
partitions that do have leaders and resume consuming the partitions that
previously had no leader once they return to normal.

Signed-off-by: liutao366 <[email protected]>
Signed-off-by: Dominic Evans <[email protected]>
Co-authored-by: Dominic Evans <[email protected]>

Author: liutao365

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)