Repository: Shopify/sarama
Pull request #1359 (2019-04-18 16:57:50)

Support timeout when fetching metadata
### Versions

Sarama Version: `v1.22.0-6-g2a49b70` (2a49b70a5c87e2dac5801f72d80fd2732c1c7b75)
Kafka Version: `1.0.2`
Go Version: `1.12`

### Configuration

What configuration values are you using for Sarama and Kafka?

Default configuration, see unit tests.

### Problem Description

Fetching metadata from an unreachable cluster using the default configuration (`30s` `DialTimeout`, `3` retries, `250ms` backoff) of **2** brokers can take more than **4 minutes**  before failing with `ErrOutOfBrokers`.

The following logs can be seen when running the provided unit test against `master` with unresponsive seed brokers (with TCP read timeout on `127.0.0.1:63168` and `127.0.0.1:63169` to simulate unreachable cluster with TCP dial timeout):
```
DEBUG=true go test -v -run=TestClientMetadataTimeout/timeout=10m0s
[sarama] 2019/04/18 09:38:00 Using random seed: 1555605480041448000
=== RUN   TestClientMetadataTimeout
=== RUN   TestClientMetadataTimeout/timeout=10m0s
[sarama] 2019/04/18 09:38:05 Initializing new client
[sarama] 2019/04/18 09:38:05 client/metadata fetching metadata for all topics from broker 127.0.0.1:63166
[sarama] 2019/04/18 09:38:05 Connected to broker at 127.0.0.1:63166 (unregistered)
[sarama] 2019/04/18 09:38:05 Successfully initialized new client
[sarama] 2019/04/18 09:38:05 Closed connection to broker 127.0.0.1:63166
[sarama] 2019/04/18 09:38:05 client/metadata fetching metadata for all topics from broker 127.0.0.1:63168
[sarama] 2019/04/18 09:38:05 Connected to broker at 127.0.0.1:63168 (unregistered)
[sarama] 2019/04/18 09:38:35 client/metadata got error from broker -1 while fetching metadata: read tcp 127.0.0.1:63170->127.0.0.1:63168: i/o timeout
[sarama] 2019/04/18 09:38:35 Closed connection to broker 127.0.0.1:63168
[sarama] 2019/04/18 09:38:35 client/metadata fetching metadata for all topics from broker 127.0.0.1:63169
[sarama] 2019/04/18 09:38:35 Connected to broker at 127.0.0.1:63169 (unregistered)
[sarama] 2019/04/18 09:39:05 client/metadata got error from broker -1 while fetching metadata: read tcp 127.0.0.1:63184->127.0.0.1:63169: i/o timeout
[sarama] 2019/04/18 09:39:05 Closed connection to broker 127.0.0.1:63169
[sarama] 2019/04/18 09:39:05 client/metadata no available broker to send metadata request to
[sarama] 2019/04/18 09:39:05 client/brokers resurrecting 2 dead seed brokers
[sarama] 2019/04/18 09:39:05 client/metadata retrying after 250ms... (3 attempts remaining)
[sarama] 2019/04/18 09:39:05 client/metadata fetching metadata for all topics from broker 127.0.0.1:63168
[sarama] 2019/04/18 09:39:05 Connected to broker at 127.0.0.1:63168 (unregistered)
[sarama] 2019/04/18 09:39:35 client/metadata got error from broker -1 while fetching metadata: read tcp 127.0.0.1:63187->127.0.0.1:63168: i/o timeout
[sarama] 2019/04/18 09:39:35 Closed connection to broker 127.0.0.1:63168
[sarama] 2019/04/18 09:39:35 client/metadata fetching metadata for all topics from broker 127.0.0.1:63169
[sarama] 2019/04/18 09:40:05 client/metadata got error from broker -1 while fetching metadata: read tcp 127.0.0.1:63193->127.0.0.1:63169: i/o timeout
[sarama] 2019/04/18 09:40:05 Closed connection to broker 127.0.0.1:63169
[sarama] 2019/04/18 09:40:05 client/metadata no available broker to send metadata request to
[sarama] 2019/04/18 09:40:05 client/brokers resurrecting 2 dead seed brokers
[sarama] 2019/04/18 09:40:05 client/metadata retrying after 250ms... (2 attempts remaining)
[sarama] 2019/04/18 09:40:05 client/metadata fetching metadata for all topics from broker 127.0.0.1:63168
[sarama] 2019/04/18 09:40:05 Connected to broker at 127.0.0.1:63168 (unregistered)
[sarama] 2019/04/18 09:40:35 client/metadata got error from broker -1 while fetching metadata: read tcp 127.0.0.1:63196->127.0.0.1:63168: i/o timeout
[sarama] 2019/04/18 09:40:35 Closed connection to broker 127.0.0.1:63168
[sarama] 2019/04/18 09:40:35 client/metadata fetching metadata for all topics from broker 127.0.0.1:63169
[sarama] 2019/04/18 09:40:35 Connected to broker at 127.0.0.1:63169 (unregistered)
[sarama] 2019/04/18 09:41:05 client/metadata got error from broker -1 while fetching metadata: read tcp 127.0.0.1:63204->127.0.0.1:63169: i/o timeout
[sarama] 2019/04/18 09:41:05 Closed connection to broker 127.0.0.1:63169
[sarama] 2019/04/18 09:41:05 client/metadata no available broker to send metadata request to
[sarama] 2019/04/18 09:41:05 client/brokers resurrecting 2 dead seed brokers
[sarama] 2019/04/18 09:41:05 client/metadata retrying after 250ms... (1 attempts remaining)
[sarama] 2019/04/18 09:41:05 client/metadata fetching metadata for all topics from broker 127.0.0.1:63168
[sarama] 2019/04/18 09:41:35 client/metadata got error from broker -1 while fetching metadata: read tcp 127.0.0.1:63211->127.0.0.1:63168: i/o timeout
[sarama] 2019/04/18 09:41:35 Closed connection to broker 127.0.0.1:63168
[sarama] 2019/04/18 09:41:35 client/metadata fetching metadata for all topics from broker 127.0.0.1:63169
[sarama] 2019/04/18 09:42:05 client/metadata got error from broker -1 while fetching metadata: read tcp 127.0.0.1:63230->127.0.0.1:63169: i/o timeout
[sarama] 2019/04/18 09:42:05 Closed connection to broker 127.0.0.1:63169
[sarama] 2019/04/18 09:42:05 client/metadata no available broker to send metadata request to
[sarama] 2019/04/18 09:42:05 client/brokers resurrecting 2 dead seed brokers
[sarama] 2019/04/18 09:42:05 Closing Client
    --- PASS: TestClientMetadataTimeout/timeout=10m0s (240.80s)
        client_test.go:670: Got err: kafka: client has run out of available brokers to talk to (Is your cluster reachable?) after waiting for: 4m0.792878004s
```

This time grows exponentially with the number of brokers because of the following formula:
```
Net.[Dial|Read]Timeout * BrokerCount * (Metadata.Retry.Max + 1) + Metadata.Retry.Backoff * Metadata.Retry.Max
```

### Solution

The proposed solution is to add a new configuration option `Metadata.Timeout` to fail faster.

### Changes

- add `Metadata.Timeout` configuration option to fail faster (disabled by default for backward compatibility)
- stop retry to query metadata on next broker when reaching such timeout has expired and return error
- add unit tests to validate behaviour when such configuration option is set to non `0`

### Testing done

See provided unit tests to see `ErrOutOfBrokers` returned much faster.

We have been using that approach in production successfully (based on a different Sarama release) to failover a producer from one Kafka cluster to another.
When one cluster become unreachable over WAN we can failover to another region in less than 30 seconds using the following configuration:
```go
config.Net.DialTimeout = 10 * time.Second
config.Net.ReadTimeout = 15 * time.Second
config.Net.WriteTimeout = 10 * time.Second
// Use a single retry for querying metadata rather than 3
config.Metadata.Retry.Max = 1
// Same for produce request
config.Producer.Retry.Max = 1
// Try to connect to at most two unreachable brokers
config.Metadata.Timeout = 15 * time.Second
// Disable producer retries
saramaConfig.Producer.Retry.Max = 0
// Also disable backoff as we could sleep for 100ms for every queued message
// that need to be retried if a cluster is not reachable...
saramaConfig.Producer.Retry.Backoff = 0
```

slaunay: I just signed the CLA.

Patch:
diff --git a/client.go b/client.go
index 10c41f0..0ac070c 100644
--- a/client.go
+++ b/client.go
@@ -435,7 +435,11 @@ func (client *client) RefreshMetadata(topics ...string) error {
 		}
 	}
 
-	return client.tryRefreshMetadata(topics, client.conf.Metadata.Retry.Max)
+	deadline := time.Time{}
+	if client.conf.Metadata.Timeout > 0 {
+		deadline = time.Now().Add(client.conf.Metadata.Timeout)
+	}
+	return client.tryRefreshMetadata(topics, client.conf.Metadata.Retry.Max, deadline)
 }
 
 func (client *client) GetOffset(topic string, partitionID int32, time int64) (int64, error) {
@@ -737,20 +741,32 @@ func (client *client) refreshMetadata() error {
 	return nil
 }
 
-func (client *client) tryRefreshMetadata(topics []string, attemptsRemaining int) error {
+func (client *client) tryRefreshMetadata(topics []string, attemptsRemaining int, deadline time.Time) error {
+	pastDeadline := func(backoff time.Duration) bool {
+		if !deadline.IsZero() && time.Now().Add(backoff).After(deadline) {
+			// we are past the deadline
+			return true
+		}
+		return false
+	}
 	retry := func(err error) error {
 		if attemptsRemaining > 0 {
 			backoff := client.computeBackoff(attemptsRemaining)
+			if pastDeadline(backoff) {
+				Logger.Println("client/metadata skipping last retries as we would go past the metadata timeout")
+				return err
+			}
 			Logger.Printf("client/metadata retrying after %dms... (%d attempts remaining)\n", client.conf.Metadata.Retry.Backoff/time.Millisecond, attemptsRemaining)
 			if backoff > 0 {
 				time.Sleep(backoff)
 			}
-			return client.tryRefreshMetadata(topics, attemptsRemaining-1)
+			return client.tryRefreshMetadata(topics, attemptsRemaining-1, deadline)
 		}
 		return err
 	}
 
-	for broker := client.any(); broker != nil; broker = client.any() {
+	broker := client.any()
+	for ; broker != nil && !pastDeadline(0); broker = client.any() {
 		allowAutoTopicCreation := true
 		if len(topics) > 0 {
 			Logger.Printf("client/metadata fetching metadata for %v from broker %s\n", topics, broker.addr)
@@ -788,6 +804,11 @@ func (client *client) tryRefreshMetadata(topics []string, attemptsRemaining int)
 		}
 	}
 
+	if broker != nil {
+		Logger.Println("client/metadata not fetching metadata from broker %s as we would go past the metadata timeout\n", broker.addr)
+		return retry(ErrOutOfBrokers)
+	}
+
 	Logger.Println("client/metadata no available broker to send metadata request to")
 	client.resurrectDeadBrokers()
 	return retry(ErrOutOfBrokers)
diff --git a/config.go b/config.go
index d0cfa62..9cba935 100644
--- a/config.go
+++ b/config.go
@@ -121,6 +121,13 @@ type Config struct {
 		// and usually more convenient, but can take up a substantial amount of
 		// memory if you have many topics and partitions. Defaults to true.
 		Full bool
+
+		// How long to wait for a successful metadata response.
+		// Disabled by default which means a metadata request against an unreachable
+		// cluster (all brokers are unreachable or unresponsive) can take up to
+		// `Net.[Dial|Read]Timeout * BrokerCount * (Metadata.Retry.Max + 1) + Metadata.Retry.Backoff * Metadata.Retry.Max`
+		// to fail.
+		Timeout time.Duration
 	}
 
 	// Producer is the namespace for configuration related to producing messages,


Test patch:
diff --git a/client_test.go b/client_test.go
index d0290bb..86fecac 100644
--- a/client_test.go
+++ b/client_test.go
@@ -1,6 +1,7 @@
 package sarama
 
 import (
+	"fmt"
 	"io"
 	"sync"
 	"sync/atomic"
@@ -612,6 +613,75 @@ func TestClientController(t *testing.T) {
 	}
 }
 
+func TestClientMetadataTimeout(t *testing.T) {
+	for _, timeout := range []time.Duration{
+		250 * time.Millisecond, // Will cut the first retry pass
+		500 * time.Millisecond, // Will cut the second retry pass
+		750 * time.Millisecond, // Will cut the third retry pass
+		900 * time.Millisecond, // Will stop after the three retries
+	} {
+		t.Run(fmt.Sprintf("timeout=%v", timeout), func(t *testing.T) {
+			// Use a responsive broker to create a working client
+			initialSeed := NewMockBroker(t, 0)
+			emptyMetadata := new(MetadataResponse)
+			initialSeed.Returns(emptyMetadata)
+
+			conf := NewConfig()
+			// Speed up the metadata request failure because of a read timeout
+			conf.Net.ReadTimeout = 100 * time.Millisecond
+			// Disable backoff and refresh
+			conf.Metadata.Retry.Backoff = 0
+			conf.Metadata.RefreshFrequency = 0
+			// But configure a "global" timeout
+			conf.Metadata.Timeout = timeout
+			c, err := NewClient([]string{initialSeed.Addr()}, conf)
+			if err != nil {
+				t.Fatal(err)
+			}
+			initialSeed.Close()
+
+			client := c.(*client)
+
+			// Start seed brokers that do not reply to anything and therefore a read
+			// on the TCP connection will timeout to simulate unresponsive brokers
+			seed1 := NewMockBroker(t, 1)
+			defer seed1.Close()
+			seed2 := NewMockBroker(t, 2)
+			defer seed2.Close()
+
+			// Overwrite the seed brokers with a fixed ordering to make this test deterministic
+			safeClose(t, client.seedBrokers[0])
+			client.seedBrokers = []*Broker{NewBroker(seed1.Addr()), NewBroker(seed2.Addr())}
+			client.deadSeeds = []*Broker{}
+
+			// Start refreshing metadata in the background
+			errChan := make(chan error)
+			start := time.Now()
+			go func() {
+				errChan <- c.RefreshMetadata()
+			}()
+
+			// Check that the refresh fails fast enough (less than twice the configured timeout)
+			// instead of at least: 100 ms * 2 brokers * 3 retries = 800 ms
+			maxRefreshDuration := 2 * timeout
+			select {
+			case err := <-errChan:
+				t.Logf("Got err: %v after waiting for: %v", err, time.Since(start))
+				if err == nil {
+					t.Fatal("Expected failed RefreshMetadata, got nil")
+				}
+				if err != ErrOutOfBrokers {
+					t.Error("Expected failed RefreshMetadata with ErrOutOfBrokers, got:", err)
+				}
+			case <-time.After(maxRefreshDuration):
+				t.Fatalf("RefreshMetadata did not fail fast enough after waiting for %v", maxRefreshDuration)
+			}
+
+			safeClose(t, c)
+		})
+	}
+}
+
 func TestClientCoordinatorWithConsumerOffsetsTopic(t *testing.T) {
 	seedBroker := NewMockBroker(t, 1)
 	staleCoordinator := NewMockBroker(t, 2)