fix: clear preferredReadReplica if broker shutdown

After Sarama had been given a preferred replica to consume from, it was
mistakenly latching onto that value and not unsetting it in the case
that the preferred replica broker was shutdown and left the cluster
metadata.

Fetches continued to work as long as that broker remained shutdown,
because they were now being sent to the Leader, which would service them
itself as it had no better preferred replica to point the client at.

However, consumption would then hang after the broker came back online,
because the Leader would stop returning records in the FetchResponse and
would instead just return the preferred replicaID, expecting the
client to send its FetchRequests over there. However, because the
partitionConsumer had latched the value of preferredReplica it never
dispatched to (re-)connect to the preferred replica and instead just
continued to send FetchRequests to the leader and received no records
back.

Contributes-to: #2090

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

Author: dnwe

consumer.go

@@ -351,7 +351,6 @@ func (child *partitionConsumer) dispatcher() {
 				child.broker = nil
 			}
 
-			Logger.Printf("consumer/%s/%d finding new broker\n", child.topic, child.partition)
 			if err := child.dispatch(); err != nil {
 				child.sendError(err)
 				child.trigger <- none{}
@@ -372,6 +371,14 @@ func (child *partitionConsumer) preferredBroker() (*Broker, error) {
 		if err == nil {
 			return broker, nil
 		}
+		Logger.Printf(
+			"consumer/%s/%d failed to find active broker for preferred read replica %d - will fallback to leader",
+			child.topic, child.partition, child.preferredReadReplica)
+
+		// if we couldn't find it, discard the replica preference and trigger a
+		// metadata refresh whilst falling back to consuming from the leader again
+		child.preferredReadReplica = invalidPreferredReplicaID
+		_ = child.consumer.client.RefreshMetadata(child.topic)
 	}
 
 	// if preferred replica cannot be found fallback to leader
@@ -856,6 +863,9 @@ func (bc *brokerConsumer) handleResponses() {
 			if preferredBroker, err := child.preferredBroker(); err == nil {
 				if bc.broker.ID() != preferredBroker.ID() {
 					// not an error but needs redispatching to consume from preferred replica
+					Logger.Printf(
+						"consumer/broker/%d abandoned in favour of preferred replica broker/%d\n",
+						bc.broker.ID(), preferredBroker.ID())
 					child.trigger <- none{}
 					delete(bc.subscriptions, child)
 				}
@@ -864,7 +874,7 @@ func (bc *brokerConsumer) handleResponses() {
 		}
 
 		// Discard any replica preference.
-		child.preferredReadReplica = -1
+		child.preferredReadReplica = invalidPreferredReplicaID
 
 		switch result {
 		case errTimedOut:

functional_consumer_follower_fetch_test.go

@@ -0,0 +1,127 @@
+//go:build functional
+// +build functional
+
+package sarama
+
+import (
+	"context"
+	"fmt"
+	"strconv"
+	"sync"
+	"testing"
+	"time"
+)
+
+func TestConsumerFetchFollowerFailover(t *testing.T) {
+	const (
+		topic  = "test.1"
+		numMsg = 1000
+	)
+
+	newConfig := func() *Config {
+		config := NewConfig()
+		config.ClientID = t.Name()
+		config.Version = V2_8_0_0
+		config.Producer.Return.Successes = true
+		return config
+	}
+
+	config := newConfig()
+
+	// pick a partition and find the ID for one of the follower brokers
+	admin, err := NewClusterAdmin(FunctionalTestEnv.KafkaBrokerAddrs, config)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer admin.Close()
+
+	metadata, err := admin.DescribeTopics([]string{topic})
+	if err != nil {
+		t.Fatal(err)
+	}
+	partition := metadata[0].Partitions[0]
+	leader := metadata[0].Partitions[0].Leader
+	follower := int32(-1)
+	for _, replica := range partition.Replicas {
+		if replica == leader {
+			continue
+		}
+		follower = replica
+		break
+	}
+
+	t.Logf("topic %s has leader kafka-%d and our chosen follower is kafka-%d", topic, leader, follower)
+
+	// match our clientID to the given broker so our requests should end up fetching from that follower
+	config.RackID = strconv.FormatInt(int64(follower), 10)
+
+	consumer, err := NewConsumer(FunctionalTestEnv.KafkaBrokerAddrs, config)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	pc, err := consumer.ConsumePartition(topic, partition.ID, OffsetOldest)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() {
+		pc.Close()
+		consumer.Close()
+	}()
+
+	producer, err := NewSyncProducer(FunctionalTestEnv.KafkaBrokerAddrs, config)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer producer.Close()
+
+	var wg sync.WaitGroup
+	wg.Add(numMsg)
+
+	go func() {
+		for i := 0; i < numMsg; i++ {
+			msg := &ProducerMessage{
+				Topic: topic, Key: nil, Value: StringEncoder(fmt.Sprintf("%s %-3d", t.Name(), i))}
+			if _, offset, err := producer.SendMessage(msg); err != nil {
+				t.Error(i, err)
+			} else if offset%50 == 0 {
+				t.Logf("sent: %d\n", offset)
+			}
+			wg.Done()
+			time.Sleep(time.Millisecond * 25)
+		}
+	}()
+
+	i := 0
+
+	for ; i < numMsg/8; i++ {
+		msg := <-pc.Messages()
+		if msg.Offset%50 == 0 {
+			t.Logf("recv: %d\n", msg.Offset)
+		}
+	}
+
+	if err := stopDockerTestBroker(context.Background(), follower); err != nil {
+		t.Fatal(err)
+	}
+
+	for ; i < numMsg/3; i++ {
+		msg := <-pc.Messages()
+		if msg.Offset%50 == 0 {
+			t.Logf("recv: %d\n", msg.Offset)
+		}
+	}
+
+	if err := startDockerTestBroker(context.Background(), follower); err != nil {
+		t.Fatal(err)
+	}
+
+	for ; i < numMsg; i++ {
+		msg := <-pc.Messages()
+		if msg.Offset%50 == 0 {
+			t.Logf("recv: %d\n", msg.Offset)
+		}
+	}
+
+	wg.Wait()
+}

functional_test.go

@@ -221,6 +221,7 @@ func existingEnvironment(ctx context.Context, env *testEnvironment) (bool, error
 	toxiproxyHost := toxiproxyURL.Hostname()
 
 	env.ToxiproxyClient = toxiproxy.NewClient(toxiproxyAddr)
+	env.Proxies = map[string]*toxiproxy.Proxy{}
 	for i := 1; i <= 5; i++ {
 		proxyName := fmt.Sprintf("kafka%d", i)
 		proxy, err := env.ToxiproxyClient.Proxy(proxyName)
@@ -262,6 +263,26 @@ func tearDownDockerTestEnvironment(ctx context.Context, env *testEnvironment) er
 	return nil
 }
 
+func startDockerTestBroker(ctx context.Context, brokerID int32) error {
+	c := exec.Command("docker-compose", "start", fmt.Sprintf("kafka-%d", brokerID))
+	c.Stdout = os.Stdout
+	c.Stderr = os.Stderr
+	if err := c.Run(); err != nil {
+		return fmt.Errorf("failed to run docker-compose to start test broker kafka-%d: %w", brokerID, err)
+	}
+	return nil
+}
+
+func stopDockerTestBroker(ctx context.Context, brokerID int32) error {
+	c := exec.Command("docker-compose", "stop", fmt.Sprintf("kafka-%d", brokerID))
+	c.Stdout = os.Stdout
+	c.Stderr = os.Stderr
+	if err := c.Run(); err != nil {
+		return fmt.Errorf("failed to run docker-compose to stop test broker kafka-%d: %w", brokerID, err)
+	}
+	return nil
+}
+
 func prepareTestTopics(ctx context.Context, env *testEnvironment) error {
 	Logger.Println("creating test topics")
 	var testTopicNames []string