fix(consumer): add recovery from no leader partitions

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]>

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,11 @@ type nopCloserClient struct {
 func (ncc *nopCloserClient) Close() error {
 	return nil
 }
+
+func (client *client) PartitionNotReadable(topic string, partition int32) bool {
+	pm := client.metadata[topic][partition]
+	if pm == nil {
+		return true
+	}
+	return pm.Leader == -1
+}

consumer_group.go

@@ -864,6 +864,28 @@ func newConsumerGroupSession(ctx context.Context, parent *consumerGroup, claims
 	// start consuming
 	for topic, partitions := range claims {
 		for _, partition := range partitions {
+			if parent.client.PartitionNotReadable(topic, partition) {
+				// partition not readable, wait for it to become readable
+				go func(topic string, partition int32) {
+					timer := time.NewTimer(5 * time.Second)
+					for parent.client.PartitionNotReadable(topic, partition) {
+						select {
+						case <-ctx.Done():
+							return
+						case <-parent.closed:
+							return
+						case <-timer.C:
+							timer.Reset(5 * time.Second)
+						}
+					}
+					timer.Stop()
+					sess.waitGroup.Add(1)
+					defer sess.waitGroup.Done()
+					defer sess.cancel()
+					sess.consume(topic, partition)
+				}(topic, partition)
+				continue
+			}
 			sess.waitGroup.Add(1)
 
 			go func(topic string, partition int32) {