Merge pull request #2 from vend/retry

Add connection retry logic

cmd/createTopic.go

@@ -18,18 +18,11 @@ kafkaCLI createTopic --bootstrap-server kafka:9092 --partitions 4 --replication-
 `,
 	Args: cobra.MinimumNArgs(1),
 	Run: func(cmd *cobra.Command, args []string) {
-		client, err := kafkaClient()
-		if err != nil {
-			panic(err)
-		}
-		kafkaAdmin, err := sarama.NewClusterAdminFromClient(client)
-		if err != nil {
-			panic(err)
-		}
+		admin := kafkaAdmin()
 
 		for _, topicName := range args {
 			fmt.Println("Creating topic " + topicName)
-			err = kafkaAdmin.CreateTopic(topicName, topicDetail(), false)
+			err := admin.CreateTopic(topicName, topicDetail(), false)
 			if err != nil {
 				switch err.(type) {
 				case *sarama.TopicError:
@@ -51,7 +44,7 @@ kafkaCLI createTopic --bootstrap-server kafka:9092 --partitions 4 --replication-
 			}
 		}
 
-		_ = kafkaAdmin.Close()
+		_ = admin.Close()
 	},
 }
 

cmd/deleteTopic.go

@@ -17,18 +17,11 @@ kafkaCLI deleteTopic --bootstrap-server kafka:9092 topic1 topic2
 `,
 	Args: cobra.MinimumNArgs(1),
 	Run: func(cmd *cobra.Command, args []string) {
-		client, err := kafkaClient()
-		if err != nil {
-			panic(err)
-		}
-		kafkaAdmin, err := sarama.NewClusterAdminFromClient(client)
-		if err != nil {
-			panic(err)
-		}
+		admin := kafkaAdmin()
 
 		for _, topicName := range args {
 			fmt.Println("Deleting topic " + topicName)
-			err = kafkaAdmin.DeleteTopic(topicName)
+			err := admin.DeleteTopic(topicName)
 
 			if err != nil {
 				switch err {
@@ -40,6 +33,8 @@ kafkaCLI deleteTopic --bootstrap-server kafka:9092 topic1 topic2
 				}
 			}
 		}
+
+		_ = admin.Close()
 	},
 }
 

cmd/root.go

@@ -3,6 +3,7 @@ package cmd
 import (
 	"fmt"
 	"os"
+	"time"
 
 	"github.com/Shopify/sarama"
 	"github.com/spf13/cobra"
@@ -30,9 +31,50 @@ func init() {
 	_ = rootCmd.MarkPersistentFlagRequired("bootstrap-server")
 }
 
-func kafkaClient() (sarama.Client, error) {
+func kafkaClient() sarama.Client {
 	kafkaConfig := sarama.NewConfig()
 	kafkaConfig.Version = sarama.V1_0_0_0
 	addresses := []string{bootstrapServer}
-	return sarama.NewClient(addresses, kafkaConfig)
+
+	var client sarama.Client
+	var err error
+
+	for i := 0; i < 30; i++ {
+		client, err = sarama.NewClient(addresses, kafkaConfig)
+		if err == nil {
+			break
+		}
+
+		fmt.Println("failed to connect to " + bootstrapServer + " Retrying in 1s")
+		time.Sleep(time.Second)
+	}
+
+	if err != nil {
+		panic(err)
+	}
+
+	return client
+}
+
+func kafkaAdmin() sarama.ClusterAdmin {
+	client := kafkaClient()
+
+	var admin sarama.ClusterAdmin
+	var err error
+
+	for i := 0; i < 10; i++ {
+		admin, err = sarama.NewClusterAdminFromClient(client)
+		if err == nil {
+			break
+		}
+
+		fmt.Println("failed to admin cluster at " + bootstrapServer + " Retrying in 1s")
+		time.Sleep(time.Second)
+	}
+
+	if err != nil {
+		panic(err)
+	}
+
+	return admin
 }