add topic update cmd with support for partitions

birdayz · Apr 8, 2020 · 4fd7f94 · 4fd7f94
1 parent fb5467c
commit 4fd7f94
Show file tree

Hide file tree

Showing 2 changed files with 38 additions and 45 deletions.
diff --git a/cmd/kaf/topic.go b/cmd/kaf/topic.go
@@ -8,15 +8,18 @@ import (
 
 	"strings"
 
+	"encoding/json"
+
 	"github.com/Shopify/sarama"
 	"github.com/spf13/cobra"
 )
 
 var (
-	partitionsFlag int32
-	replicasFlag   int16
-	noHeaderFlag   bool
-	compactFlag    bool
+	partitionsFlag           int32
+	partitionAssignmentsFlag string
+	replicasFlag             int16
+	noHeaderFlag             bool
+	compactFlag              bool
 )
 
 func init() {
@@ -28,13 +31,16 @@ func init() {
 	topicCmd.AddCommand(describeTopicCmd)
 	topicCmd.AddCommand(addConfigCmd)
 	topicCmd.AddCommand(topicSetConfig)
+	topicCmd.AddCommand(updateTopicCmd)
 
 	createTopicCmd.Flags().Int32VarP(&partitionsFlag, "partitions", "p", int32(1), "Number of partitions")
 	createTopicCmd.Flags().Int16VarP(&replicasFlag, "replicas", "r", int16(1), "Number of replicas")
 	createTopicCmd.Flags().BoolVar(&compactFlag, "compact", false, "Enable topic compaction")
 
 	lsTopicsCmd.Flags().BoolVar(&noHeaderFlag, "no-headers", false, "Hide table headers")
 	topicsCmd.Flags().BoolVar(&noHeaderFlag, "no-headers", false, "Hide table headers")
+	updateTopicCmd.Flags().Int32VarP(&partitionsFlag, "partitions", "p", int32(-1), "Number of partitions")
+	updateTopicCmd.Flags().StringVar(&partitionAssignmentsFlag, "partition-assignments", "", "Partition Assignments. Optional. If set, an assignment must be provided for each new partition. Example: '[[1,2,3],[1,2,3]]' (JSON Array syntax) assigns two new partitions to brokers 1,2,3")
 }
 
 var topicCmd = &cobra.Command{
@@ -85,6 +91,34 @@ var topicSetConfig = &cobra.Command{
 	},
 }
 
+var updateTopicCmd = &cobra.Command{
+	Use:     "update",
+	Short:   "Update topic",
+	Example: "kaf topic update -p 5 --partition-assignments '[[1,2,3],[1,2,3]]'",
+	Args:    cobra.ExactArgs(1),
+	Run: func(cmd *cobra.Command, args []string) {
+		admin := getClusterAdmin()
+
+		if partitionsFlag == -1 {
+			errorExit("Number of partitions must be given")
+		}
+
+		var assignments [][]int32
+		if partitionAssignmentsFlag != "" {
+			if err := json.Unmarshal([]byte(partitionAssignmentsFlag), &assignments); err != nil {
+				errorExit("Invalid partition assignments: %v", err)
+			}
+		}
+
+		err := admin.CreatePartitions(args[0], partitionsFlag, assignments, false)
+		if err != nil {
+			errorExit("Failed to create partitions: %v", err)
+		}
+
+		fmt.Printf("\xE2\x9C\x85 Updated topic!\n")
+	},
+}
+
 var lsTopicsCmd = &cobra.Command{
 	Use:     "ls",
 	Aliases: []string{"list"},