Permit Metadata.Retry.Backoff == 0

We permit it everywhere else we have backoffs. Use it to test the client's retry
strategy.

client_test.go

@@ -206,7 +206,8 @@ func TestClientReceivingUnknownTopic(t *testing.T) {
 	seedBroker.Returns(metadataResponse1)
 
 	config := NewConfig()
-	config.Metadata.Retry.Max = 0
+	config.Metadata.Retry.Max = 1
+	config.Metadata.Retry.Backoff = 0
 	client, err := NewClient([]string{seedBroker.Addr()}, config)
 	if err != nil {
 		t.Fatal(err)
@@ -215,6 +216,7 @@ func TestClientReceivingUnknownTopic(t *testing.T) {
 	metadataUnknownTopic := new(MetadataResponse)
 	metadataUnknownTopic.AddTopic("new_topic", ErrUnknownTopicOrPartition)
 	seedBroker.Returns(metadataUnknownTopic)
+	seedBroker.Returns(metadataUnknownTopic)
 
 	if err := client.RefreshMetadata("new_topic"); err != ErrUnknownTopicOrPartition {
 		t.Error("ErrUnknownTopicOrPartition expected, got", err)
@@ -223,6 +225,7 @@ func TestClientReceivingUnknownTopic(t *testing.T) {
 	// If we are asking for the leader of a partition of the non-existing topic.
 	// we will request metadata again.
 	seedBroker.Returns(metadataUnknownTopic)
+	seedBroker.Returns(metadataUnknownTopic)
 
 	if _, err = client.Leader("new_topic", 1); err != ErrUnknownTopicOrPartition {
 		t.Error("Expected ErrUnknownTopicOrPartition, got", err)

config.go

@@ -192,9 +192,9 @@ func (c *Config) Validate() error {
 	switch {
 	case c.Metadata.Retry.Max < 0:
 		return ConfigurationError("Invalid Metadata.Retry.Max, must be >= 0")
-	case c.Metadata.Retry.Backoff <= time.Duration(0):
-		return ConfigurationError("Invalid Metadata.Retry.Backoff, must be > 0")
-	case c.Metadata.RefreshFrequency < time.Duration(0):
+	case c.Metadata.Retry.Backoff < 0:
+		return ConfigurationError("Invalid Metadata.Retry.Backoff, must be >= 0")
+	case c.Metadata.RefreshFrequency < 0:
 		return ConfigurationError("Invalid Metadata.RefreshFrequency, must be >= 0")
 	}