From 407fc37608c289a01d434447c6a113e4b4b090d7 Mon Sep 17 00:00:00 2001 From: Conor Mongey Date: Sun, 4 Feb 2018 06:19:25 +0000 Subject: [PATCH] Clear all metadata when we have the latest info --- client.go | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/client.go b/client.go index 937000a7b9..f5563e9384 100644 --- a/client.go +++ b/client.go @@ -649,8 +649,9 @@ func (client *client) tryRefreshMetadata(topics []string, attemptsRemaining int) switch err.(type) { case nil: + allKnownMetaData := len(topics) == 0 // valid response, use it - shouldRetry, err := client.updateMetadata(response) + shouldRetry, err := client.updateMetadata(response, allKnownMetaData) if shouldRetry { Logger.Println("client/metadata found some partitions to be leaderless") return retry(err) // note: err can be nil @@ -674,7 +675,7 @@ func (client *client) tryRefreshMetadata(topics []string, attemptsRemaining int) } // if no fatal error, returns a list of topics that need retrying due to ErrLeaderNotAvailable -func (client *client) updateMetadata(data *MetadataResponse) (retry bool, err error) { +func (client *client) updateMetadata(data *MetadataResponse, allKnownMetaData bool) (retry bool, err error) { client.lock.Lock() defer client.lock.Unlock() @@ -685,7 +686,10 @@ func (client *client) updateMetadata(data *MetadataResponse) (retry bool, err er for _, broker := range data.Brokers { client.registerBroker(broker) } - + if allKnownMetaData { + client.metadata = make(map[string]map[int32]*PartitionMetadata) + client.cachedPartitionsResults = make(map[string][maxPartitionIndex][]int32) + } for _, topic := range data.Topics { delete(client.metadata, topic.Name) delete(client.cachedPartitionsResults, topic.Name)