Extra sleep when __consumer_offsets does not exist yet.

client.go

@@ -687,6 +687,15 @@ func (client *client) getConsumerMetadata(consumerGroup string, retriesRemaining
 		case ErrConsumerCoordinatorNotAvailable:
 			Logger.Printf("client/coordinator Coordinator for consumer group %s is not available.\n", consumerGroup)
 
+			// This is quite shitty, but this scenario will only happen once per cluster, because the
+			// __consumer_offsets only has to be created one time.
+			if client.metadata["__consumer_offsets"] == nil {
+				if err := client.RefreshMetadata("__consumer_offsets"); err == ErrUnknownTopicOrPartition {
+					Logger.Printf("client/coordinator The __consumer_offsets topic does not yet exist.\n", consumerGroup)
+					time.Sleep(2 * time.Second)
+				}
+			}
+
 			if retriesRemaining > 0 {
 				Logger.Printf("Retrying after %dms... (%d retries remaining)\n", client.conf.Metadata.Retry.Backoff/time.Millisecond, retriesRemaining)
 				time.Sleep(client.conf.Metadata.Retry.Backoff)