Repository: IBM/sarama

Issue:
Metadata request burst
##### Description

I work at Confluent, in the Warpstream team. Both confluent internally and its customers are using sarama to drive workloads that write Gigabytes per second of traffic to Kafka and Warpstream.
I believe that sarama can overwhelm the brokers with metadata requests when the number of brokers is high and the number of partitions they each hold is high. 

Let’s imagine the setup. We have 3 brokers, and each of them is the leader for 1000 partitions. There are also 100 producers producing to these partitions.
Let us now imagine that one broker shuts down.
Inside each producer this will happen:
1. Something will try to make the producer produce to one of the partitions that broker was the leader for, and will notice that the broker is not there anymore.
2. Sarama will detect that it is `abandoned` and the code [here](https://github.com/IBM/sarama/blob/57be9bdd0a5cd133dc27edf35560b057074b9508/async_producer.go#L662) will trigger. It will set the `brokerProducer` of the partition to `nil` which will trigger a metadata refresh.
3. Sarama will do this in turn for the 1000 partitions, triggering 1000 metadata refreshes, one for each partition.

Overall, the 100 producers will, in a very short amount of time, issue 100,000 metadata requests to the broker, so that they can know where the new leader is now.

We have customers who are facing this issue on a Warpstream cluster, where the issue is exacerbated by the fact an agent can declare itself the leader for a lot of partitions. 

We would like to fix this and we are willing to contribute a patch that
1. Allows to run a single metadata request per sarama client in parallel. If other parts of the code call `RefreshMetadata()` they will be queued after the current one finishes
2. Adds a configuration boolean that allows (as an opt-in) to enable this behavior.

Would you be open to integrating this patch ?

##### Versions

| Sarama | Kafka | Go  |
| ------ | ----- | --- |
|   1.45.2     |    4.0   |  1.24   |

##### Configuration

```go

	config := sarama.NewConfig()
	config.Version = sarama.MaxVersion
	config.Metadata.RefreshFrequency = 60 * time.Second
	config.Producer.Flush.MaxMessages = 1_000_000
	config.Producer.Flush.Frequency = 25 * time.Millisecond
	config.Producer.MaxMessageBytes = 64_000_000
	config.Producer.Flush.Bytes = 16_000_000

	config.Net.ReadTimeout = 90 * time.Second
	config.Consumer.Group.Rebalance.Timeout = 60 * time.Second
	config.Consumer.Group.Session.Timeout = 45 * time.Second
	config.Consumer.Group.Heartbeat.Interval = 5 * time.Second
	config.Consumer.MaxProcessingTime = 20 * time.Second
	config.Consumer.Fetch.Default = 50_000_000
	config.Consumer.Group.Rebalance.Strategy = sarama.NewBalanceStrategyRange()
	config.Consumer.MaxWaitTime = 10 * time.Second
	config.Producer.Partitioner = func(topic string) sarama.Partitioner {
		return sarama.NewManualPartitioner(topic)
	}

	config.Net.ReadTimeout = time.Minute
	config.Consumer.Group.Heartbeat.Interval = 5 * time.Second
```

Hints from issue comments:
@dnwe: @cupcicm thanks for raising the issue, it should hopefully be quite easy to reproduce in a unittest and assert on the metadata request count

A PR sounds like a great general improvement if you have already envisioned the fix and are happy to work on it. I wouldn’t bother with a configuration option to preserve the old behavior as I don’t think it is desirable to support blasting the cluster with metadata requests@cupcicm: I have a patch ready, I am polishing it. Will send it soon.
Would you be OK with me introducing it as an opt-in and then I will let you slowly make "true" the default, so that you can do it on your own schedule ? WDYT ?@cupcicm: I think that this PR fixes this issue: https://github.com/IBM/sarama/pull/3225

Solution patch:
diff --git a/client.go b/client.go
index 21d0400..279fedf 100644
--- a/client.go
+++ b/client.go
@@ -166,6 +166,8 @@ type client struct {
 	cachedPartitionsResults map[string][maxPartitionIndex][]int32
 
 	lock sync.RWMutex // protects access to the maps that hold cluster state.
+
+	metadataRefresh metadataRefresh
 }
 
 // NewClient creates a new Client. It connects to one of the given broker addresses
@@ -192,7 +194,6 @@ func NewClient(addrs []string, conf *Config) (Client, error) {
 			conf.Version = V1_0_0_0
 		}
 	}
-
 	client := &client{
 		conf:                    conf,
 		closer:                  make(chan none),
@@ -204,6 +205,18 @@ func NewClient(addrs []string, conf *Config) (Client, error) {
 		coordinators:            make(map[string]int32),
 		transactionCoordinators: make(map[string]int32),
 	}
+	var refresh = func(topics []string) error {
+		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)
+	}
+	if conf.Metadata.SingleFlight {
+		client.metadataRefresh = newSingleFlightRefresher(refresh)
+	} else {
+		client.metadataRefresh = refresh
+	}
 
 	if conf.Net.ResolveCanonicalBootstrapServers {
 		var err error
@@ -509,12 +522,7 @@ func (client *client) RefreshMetadata(topics ...string) error {
 			return ErrInvalidTopic // this is the error that 0.8.2 and later correctly return
 		}
 	}
-
-	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)
+	return client.metadataRefresh(topics)
 }
 
 func (client *client) GetOffset(topic string, partitionID int32, timestamp int64) (int64, error) {
diff --git a/config.go b/config.go
index 6a198dc..baeca8b 100644
--- a/config.go
+++ b/config.go
@@ -167,6 +167,14 @@ type Config struct {
 		// the broker may auto-create topics that we requested which do not already exist,
 		// if it is configured to do so (`auto.create.topics.enable` is true). Defaults to true.
 		AllowAutoTopicCreation bool
+
+		// SingleFlight controls whether to send a single metadata refresh request at a given time
+		// or whether to allow anyone to refresh the metadata concurrently.
+		// If this is set to true and the client needs to refresh the metadata from different goroutines,
+		// the requests will be batched together so that a single refresh is sent at a time.
+		// See https://github.com/IBM/sarama/issues/3224 for more details.
+		// SingleFlight defaults to false.
+		SingleFlight bool
 	}
 
 	// Producer is the namespace for configuration related to producing messages,
diff --git a/metadata.go b/metadata.go
new file mode 100644
index 0000000..20620b3
--- /dev/null
+++ b/metadata.go
@@ -0,0 +1,239 @@
+package sarama
+
+import (
+	"sync"
+)
+
+type metadataRefresh func(topics []string) error
+
+// currentRefresh makes sure sarama does not issue metadata requests
+// in parallel. If we need to refresh the metadata for a list of topics,
+// this struct will check if a refresh is already ongoing, and if so, it will
+// accumulate the list of topics to refresh in the next refresh.
+// When the current refresh is over, it will queue a new metadata refresh call
+// with the accumulated list of topics.
+type currentRefresh struct {
+	// This is the function that gets called when to refresh the metadata.
+	// It is called with the list of all topics that need to be refreshed
+	// or with nil if all topics need to be refreshed.
+	refresh func(topics []string) error
+
+	mu        sync.Mutex
+	ongoing   bool
+	topicsMap map[string]struct{}
+	topics    []string
+	allTopics bool
+	chans     []chan error
+}
+
+// addTopicsFrom adds topics from the next refresh to the current refresh.
+// You need to hold the lock to call this method.
+func (r *currentRefresh) addTopicsFrom(next *nextRefresh) {
+	if next.allTopics {
+		r.allTopics = true
+		return
+	}
+	if len(next.topics) > 0 {
+		r.addTopics(next.topics)
+	}
+}
+
+// nextRefresh holds the list of topics we will need
+// to refresh in the next refresh.
+// When a refresh is ongoing, calls to RefreshMetadata() are
+// accumulated in this struct, so that we can immediately issue another
+// refresh when the current refresh is over.
+type nextRefresh struct {
+	mu        sync.Mutex
+	topics    []string
+	allTopics bool
+}
+
+// addTopics adds topics to the refresh.
+// You need to hold the lock to call this method.
+func (r *currentRefresh) addTopics(topics []string) {
+	if len(topics) == 0 {
+		r.allTopics = true
+		return
+	}
+	for _, topic := range topics {
+		if _, ok := r.topicsMap[topic]; ok {
+			continue
+		}
+		r.topicsMap[topic] = struct{}{}
+		r.topics = append(r.topics, topic)
+	}
+}
+
+func (r *nextRefresh) addTopics(topics []string) {
+	if len(topics) == 0 {
+		r.allTopics = true
+		// All topics are requested, so we can clear the topics
+		// that were previously accumulated.
+		r.topics = r.topics[:0]
+		return
+	}
+	r.topics = append(r.topics, topics...)
+}
+
+func (r *nextRefresh) clear() {
+	r.topics = r.topics[:0]
+	r.allTopics = false
+}
+
+func (r *currentRefresh) hasTopics(topics []string) bool {
+	if len(topics) == 0 {
+		// This means that the caller wants to know if the refresh is for all topics.
+		// In this case, we return true if the refresh is for all topics, or false if it is not.
+		return r.allTopics
+	}
+	if r.allTopics {
+		return true
+	}
+	for _, topic := range topics {
+		if _, ok := r.topicsMap[topic]; !ok {
+			return false
+		}
+	}
+	return true
+}
+
+// start starts a new refresh.
+// The refresh is started in a new goroutine, and this function
+// returns a channel on which the caller can wait for the refresh
+// to complete.
+// You need to hold the lock to call this method.
+func (r *currentRefresh) start() chan error {
+	r.ongoing = true
+	ch := r.wait()
+	topics := r.topics
+	if r.allTopics {
+		topics = nil
+	}
+	go func() {
+		err := r.refresh(topics)
+		r.mu.Lock()
+		defer r.mu.Unlock()
+
+		r.ongoing = false
+		for _, ch := range r.chans {
+			ch <- err
+			close(ch)
+		}
+		r.clear()
+	}()
+	return ch
+}
+
+// clear clears the refresh state.
+// You need to hold the lock to call this method.
+func (r *currentRefresh) clear() {
+	r.topics = r.topics[:0]
+	for key := range r.topicsMap {
+		delete(r.topicsMap, key)
+	}
+	r.allTopics = false
+	r.chans = r.chans[:0]
+}
+
+// wait returns the channel on which you can wait for the refresh
+// to complete.
+// You need to hold the lock to call this method.
+func (r *currentRefresh) wait() chan error {
+	if !r.ongoing {
+		panic("waiting for a refresh that is not ongoing")
+	}
+	ch := make(chan error, 1)
+	r.chans = append(r.chans, ch)
+	return ch
+}
+
+// singleFlightMetadataRefresher helps managing metadata refreshes.
+// It makes sure a sarama client never issues more than one metadata refresh
+// in parallel.
+type singleFlightMetadataRefresher struct {
+	current *currentRefresh
+	next    *nextRefresh
+}
+
+func newSingleFlightRefresher(f func(topics []string) error) metadataRefresh {
+	return newMetadataRefresh(f).Refresh
+}
+
+func newMetadataRefresh(f func(topics []string) error) *singleFlightMetadataRefresher {
+	return &singleFlightMetadataRefresher{
+		current: &currentRefresh{
+			topicsMap: make(map[string]struct{}),
+			refresh:   f,
+		},
+		next: &nextRefresh{},
+	}
+}
+
+// Refresh is the function that clients call when they want to refresh
+// the metadata. This function blocks until a refresh is issued, and its
+// result is received, for the list of topics the caller provided.
+// If a refresh was already ongoing for this list of topics, the function
+// waits on that refresh to complete, and returns its result.
+// If a refresh was already ongoing for a different list of topics, the function
+// accumulates the list of topics to refresh in the next refresh, and queues that refresh.
+// If no refresh is ongoing, it will start a new refresh, and return its result.
+func (m *singleFlightMetadataRefresher) Refresh(topics []string) error {
+	for {
+		ch, queued := m.refreshOrQueue(topics)
+		if !queued {
+			return <-ch
+		}
+		<-ch
+	}
+}
+
+// refreshOrQueue returns a channel the refresh needs to wait on, and a boolean
+// that indicates whether waiting on the channel will return the result of that refresh
+// or whether the refresh was "queued" and the caller needs to wait for the channel to
+// return, and then call refreshOrQueue again.
+// When calling refreshOrQueue, three things can happen:
+//  1. either no refresh is ongoing.
+//     In this case, a new refresh is started, and the channel that's returned will
+//     contain the result of that refresh, so it returns "false" as the second return value.
+//  2. a refresh is ongoing, and it contains the topics we need.
+//     In this case, the channel that's returned will contain the result of that refresh,
+//     so it returns "false" as the second return value.
+//     In this case, the channel that's returned will contain the result of that refresh,
+//     so it returns "false" as the second return value.
+//  3. a refresh is already ongoing, but doesn't contain the topics we need. In this case,
+//     the caller needs to wait for the refresh to finish, and then call refreshOrQueue again.
+//     The channel that's returned is for the current refresh (not the one the caller is
+//     interested in), so it returns "true" as the second return value. The caller needs to
+//     wait on the channel, disregard the value, and call refreshOrQueue again.
+func (m *singleFlightMetadataRefresher) refreshOrQueue(topics []string) (chan error, bool) {
+	m.current.mu.Lock()
+	defer m.current.mu.Unlock()
+	if !m.current.ongoing {
+		// If no refresh is ongoing, we can start a new one, in which
+		// we add the topics that have been accumulated in the next refresh
+		// and the topics that have been provided by the caller.
+		m.next.mu.Lock()
+		m.current.addTopicsFrom(m.next)
+		m.next.clear()
+		m.next.mu.Unlock()
+		m.current.addTopics(topics)
+		ch := m.current.start()
+		return ch, false
+	}
+	if m.current.hasTopics(topics) {
+		// A refresh is ongoing, and we were lucky: it is refreshing the topics we need already:
+		// we just have to wait for it to finish and return its results.
+		ch := m.current.wait()
+		return ch, false
+	}
+	// There is a refresh ongoing, but it is not refreshing the topics we need.
+	// We need to wait for it to finish, and then start a new refresh.
+	ch := m.current.wait()
+	m.next.mu.Lock()
+	m.next.addTopics(topics)
+	m.next.mu.Unlock()
+	// This is where we wait for that refresh to finish, and the loop will take care
+	// of starting the new one.
+	return ch, true
+}



Test patch:
diff --git a/client_test.go b/client_test.go
index 7d1689e..9d6242f 100644
--- a/client_test.go
+++ b/client_test.go
@@ -4,6 +4,7 @@ package sarama
 
 import (
 	"errors"
+	"fmt"
 	"io"
 	"sync"
 	"sync/atomic"
@@ -12,6 +13,7 @@ import (
 	"time"
 
 	"github.com/rcrowley/go-metrics"
+	"github.com/stretchr/testify/assert"
 )
 
 func TestSimpleClient(t *testing.T) {
@@ -813,74 +815,94 @@ func TestClientMetadataTimeout(t *testing.T) {
 		},
 	}
 
-	for _, tc := range tests {
-		t.Run(tc.name, func(t *testing.T) {
-			// Use a responsive broker to create a working client
-			initialSeed := NewMockBroker(t, 0)
-			emptyMetadata := new(MetadataResponse)
-			emptyMetadata.AddBroker(initialSeed.Addr(), initialSeed.BrokerID())
-			initialSeed.Returns(emptyMetadata)
-
-			conf := NewTestConfig()
-			// 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 = tc.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)
-			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 * tc.timeout
-			select {
-			case err := <-errChan:
-				if err == nil {
-					t.Fatal("Expected failed RefreshMetadata, got nil")
+	for _, singleFlight := range []bool{true, false} {
+		for _, tc := range tests {
+			t.Run(fmt.Sprintf("%s_singleflight_is_%t", tc.name, singleFlight), func(t *testing.T) {
+				// Use a responsive broker to create a working client
+				initialSeed := NewMockBroker(t, 0)
+				emptyMetadata := new(MetadataResponse)
+				emptyMetadata.AddBroker(initialSeed.Addr(), initialSeed.BrokerID())
+				initialSeed.Returns(emptyMetadata)
+
+				conf := NewTestConfig()
+				// 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 = tc.timeout
+				conf.Metadata.SingleFlight = singleFlight
+				c, err := NewClient([]string{initialSeed.Addr()}, conf)
+				if err != nil {
+					t.Fatal(err)
 				}
-				if !errors.Is(err, ErrOutOfBrokers) {
-					t.Error("Expected failed RefreshMetadata with ErrOutOfBrokers, got:", 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)
+				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 * tc.timeout
+				select {
+				case err := <-errChan:
+					if err == nil {
+						t.Fatal("Expected failed RefreshMetadata, got nil")
+					}
+					if !errors.Is(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)
 				}
-			case <-time.After(maxRefreshDuration):
-				t.Fatalf("RefreshMetadata did not fail fast enough after waiting for %v", maxRefreshDuration)
-			}
 
-			safeClose(t, c)
-		})
+				safeClose(t, c)
+			})
+		}
 	}
 }
 
 func TestClientUpdateMetadataErrorAndRetry(t *testing.T) {
 	seedBroker := NewMockBroker(t, 1)
+	var called atomic.Int32
 
-	metadataResponse1 := new(MetadataResponse)
-	metadataResponse1.AddBroker(seedBroker.Addr(), 1)
-	seedBroker.Returns(metadataResponse1)
+	seedBroker.setHandler(func(req *request) (res encoderWithHeader) {
+		if req.body.key() != 3 {
+			t.Error("this test sends only Metadata requests")
+			return
+		}
+		resp := new(MetadataResponse)
+		for _, topic := range req.body.(*MetadataRequest).Topics {
+			if topic == "new_topic" {
+				resp.Topics = append(resp.Topics, &TopicMetadata{
+					Version: 1,
+					Name:    "new_topic",
+					Err:     ErrUnknownTopicOrPartition,
+				})
+			}
+		}
+		called.Add(1)
+		resp.AddBroker(seedBroker.Addr(), 1)
+		return resp
+	})
 
 	config := NewTestConfig()
 	config.Metadata.Retry.Max = 3
@@ -888,6 +910,7 @@ func TestClientUpdateMetadataErrorAndRetry(t *testing.T) {
 	config.Metadata.RefreshFrequency = 0
 	config.Net.ReadTimeout = 10 * time.Millisecond
 	config.Net.WriteTimeout = 10 * time.Millisecond
+	config.Metadata.SingleFlight = true
 	client, err := NewClient([]string{seedBroker.Addr()}, config)
 	if err != nil {
 		t.Fatal(err)
@@ -897,11 +920,7 @@ func TestClientUpdateMetadataErrorAndRetry(t *testing.T) {
 	for i := 0; i < 10; i++ {
 		go func() {
 			defer waitGroup.Done()
-			var failedMetadataResponse MetadataResponse
-			failedMetadataResponse.AddBroker(seedBroker.Addr(), 1)
-			failedMetadataResponse.AddTopic("new_topic", ErrUnknownTopicOrPartition)
-			seedBroker.Returns(&failedMetadataResponse)
-			err := client.RefreshMetadata()
+			err := client.RefreshMetadata("new_topic")
 			if err == nil {
 				t.Error("should return error")
 				return
@@ -911,6 +930,74 @@ func TestClientUpdateMetadataErrorAndRetry(t *testing.T) {
 	waitGroup.Wait()
 	safeClose(t, client)
 	seedBroker.Close()
+	// The refresh metadata is always for the same topic,
+	// it should have been batched.
+	if count := called.Load(); count >= 7 {
+		t.Errorf("Refresh metadata was called %d times, this should be less than 7.", count)
+	}
+}
+
+func TestClientRefreshesMetadataConcurrently(t *testing.T) {
+	seedBroker := NewMockBroker(t, 1)
+
+	seedBroker.setHandler(func(req *request) (res encoderWithHeader) {
+		time.Sleep(10 * time.Millisecond)
+		if req.body.key() != 3 {
+			t.Error("this test sends only Metadata requests")
+			return
+		}
+		topics := req.body.(*MetadataRequest).Topics
+		resp := new(MetadataResponse)
+		for _, topic := range topics {
+			switch topic {
+			case "topic1":
+				resp.Topics = append(resp.Topics, &TopicMetadata{
+					Version:    1,
+					Name:       "topic1",
+					Partitions: []*PartitionMetadata{},
+				})
+			case "topic2":
+				resp.Topics = append(resp.Topics, &TopicMetadata{
+					Version:    1,
+					Name:       "topic2",
+					Partitions: []*PartitionMetadata{},
+				})
+			case "topic3":
+				resp.Topics = append(resp.Topics, &TopicMetadata{
+					Version: 1,
+					Name:    "topic3",
+					Err:     ErrUnknownTopicOrPartition,
+				})
+			default:
+				t.Errorf("unexpected topic: %s", topic)
+			}
+		}
+		resp.AddBroker(seedBroker.Addr(), 1)
+		return resp
+	})
+
+	config := NewTestConfig()
+	config.Metadata.SingleFlight = true
+	client, err := NewClient([]string{seedBroker.Addr()}, config)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var waitGroup sync.WaitGroup
+	waitGroup.Add(1000)
+	for i := 0; i < 1000; i++ {
+		go func() {
+			defer waitGroup.Done()
+			assert.NoError(t, client.RefreshMetadata("topic1"))
+			assert.NoError(t, client.RefreshMetadata("topic2"))
+			assert.Error(t, client.RefreshMetadata("topic3"))
+			topics, err := client.Topics()
+			assert.NoError(t, err)
+			assert.Len(t, topics, 2)
+		}()
+	}
+	waitGroup.Wait()
+	safeClose(t, client)
+	seedBroker.Close()
 }
 
 func TestClientCoordinatorWithConsumerOffsetsTopic(t *testing.T) {
diff --git a/metadata_test.go b/metadata_test.go
new file mode 100644
index 0000000..65a7a38
--- /dev/null
+++ b/metadata_test.go
@@ -0,0 +1,103 @@
+package sarama
+
+import (
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestMetadataRefresh(t *testing.T) {
+	stepRefresh := make(chan struct{})
+	refresh := newMetadataRefresh(func(topics []string) error {
+		<-stepRefresh
+		return nil
+	})
+
+	ch, queued := refresh.refreshOrQueue([]string{"topic1"})
+	if queued {
+		t.Errorf("It's the first call, it should not be queued")
+	}
+
+	ch2, queued := refresh.refreshOrQueue([]string{"topic2", "topic3"})
+	if !queued {
+		t.Errorf("This one is requesting different topics, it should be queued")
+	}
+
+	ch3, queued := refresh.refreshOrQueue([]string{"topic3"})
+	if !queued {
+		t.Errorf("This one is requesting the same topics as the second one, it should be queued")
+	}
+
+	ch4, queued := refresh.refreshOrQueue([]string{"topic4"})
+	if !queued {
+		t.Errorf("This one is requesting different topics, it should be queued too")
+	}
+
+	ch5, queued := refresh.refreshOrQueue([]string{"topic1"})
+	if queued {
+		t.Errorf("Same topics as the first call, piggy backing on that call, so it's not queued")
+	}
+
+	stepRefresh <- struct{}{}
+	require.NoError(t, <-ch)
+	require.NoError(t, <-ch5)
+
+	require.NoError(t, <-ch2)
+	require.NoError(t, <-ch3)
+	require.NoError(t, <-ch4)
+}
+
+func TestMetadataRefreshConcurrency(t *testing.T) {
+	var firstRefreshChans []chan error
+	var lock sync.Mutex
+	stepRefresh := make(chan struct{})
+	refresh := newMetadataRefresh(func(topics []string) error {
+		<-stepRefresh
+		return nil
+	})
+
+	ch, queued := refresh.refreshOrQueue([]string{"topic1"})
+	firstRefreshChans = append(firstRefreshChans, ch)
+	if queued {
+		t.Errorf("First call, should start a refresh")
+	}
+
+	var wg sync.WaitGroup
+	for i := 0; i < 1000; i++ {
+		wg.Add(1)
+		time.Sleep(time.Millisecond)
+		go func() {
+			defer wg.Done()
+
+			ch, refreshQueued := refresh.refreshOrQueue([]string{"topic1"})
+			if refreshQueued {
+				t.Errorf("This one should not be queued: they are all requesting the topic that's already started")
+			}
+			lock.Lock()
+			firstRefreshChans = append(firstRefreshChans, ch)
+			lock.Unlock()
+		}()
+	}
+	wg.Wait()
+	// We have now queued all the refreshes, and they're all blocked with the first one.
+	stepRefresh <- struct{}{}
+	// Now they are all finished, we can pull from the channels
+	for _, ch := range firstRefreshChans {
+		require.NoError(t, <-ch)
+	}
+
+	ch, queued = refresh.refreshOrQueue([]string{"topic2", "topic3"})
+	if queued {
+		t.Errorf("This one should not be queued: no refresh is ongoing")
+	}
+	ch2, queued := refresh.refreshOrQueue([]string{"topic3", "topic4"})
+	if !queued {
+		t.Errorf("But now there is a refresh ongoing, so this one should be queued")
+	}
+
+	stepRefresh <- struct{}{}
+	require.NoError(t, <-ch)
+	require.NoError(t, <-ch2)
+}