Allow creating topic using topic description from file

cmd/create/create-topic.go

@@ -27,6 +27,7 @@ func newCreateTopicCmd() *cobra.Command {
 	cmdCreateTopic.Flags().Int32VarP(&flags.Partitions, "partitions", "p", 1, "number of partitions")
 	cmdCreateTopic.Flags().Int16VarP(&flags.ReplicationFactor, "replication-factor", "r", -1, "replication factor")
 	cmdCreateTopic.Flags().BoolVarP(&flags.ValidateOnly, "validate-only", "v", false, "validate only")
+	cmdCreateTopic.Flags().StringVarP(&flags.File, "file", "f", "", "file with topic description")
 	cmdCreateTopic.Flags().StringArrayVarP(&flags.Configs, "config", "c", flags.Configs, "configs in format `key=value`")
 
 	return cmdCreateTopic

cmd/create/create-topic_test.go

@@ -2,6 +2,7 @@ package create_test
 
 import (
 	"fmt"
+	"os"
 	"testing"
 	"time"
 
@@ -109,6 +110,101 @@ partitions:
 	testutil.AssertEquals(t, fmt.Sprintf(expected, topicName), stdOut)
 }
 
+func TestCreateTopicWithConfigFileIntegration(t *testing.T) {
+
+	testutil.StartIntegrationTest(t)
+
+	kafkaCtl := testutil.CreateKafkaCtlCommand()
+
+	topicName := testutil.GetPrefixedName("new-topic")
+	configFile := fmt.Sprintf(`
+name: %s
+partitions:
+- id: 0
+  oldestOffset: 0
+  newestOffset: 290
+  leader: kafka:9092
+  replicas: [1]
+  inSyncReplicas: [1]
+- id: 1
+  oldestOffset: 0
+  newestOffset: 258
+  leader: kafka:9092
+  replicas: [1]
+  inSyncReplicas: [1]
+- id: 2
+  oldestOffset: 0
+  newestOffset: 290
+  leader: kafka:9092
+  replicas: [1]
+  inSyncReplicas: [1]
+configs:
+- name: cleanup.policy
+  value: compact
+- name: max.message.bytes
+  value: "10485880"
+- name: min.cleanable.dirty.ratio
+  value: "1.0E-4"
+- name: delete.retention.ms
+  value: "0"
+- name: segment.ms
+  value: "100"
+`, topicName)
+
+	tmpFile, err := os.CreateTemp("", fmt.Sprintf("%s-*.yaml", topicName))
+	if err != nil {
+		t.Fatalf("could not create temp file with topic config: %s", err)
+	}
+	defer tmpFile.Close()
+	if _, err := tmpFile.WriteString(configFile); err != nil {
+		t.Fatalf("could not write temp config file: %s", err)
+	}
+
+	if _, err := kafkaCtl.Execute("create", "topic", topicName, "-f", tmpFile.Name()); err != nil {
+		t.Fatalf("failed to execute command: %v", err)
+	}
+
+	testutil.AssertEquals(t, fmt.Sprintf("topic created: %s", topicName), kafkaCtl.GetStdOut())
+
+	describeTopic(t, kafkaCtl, topicName)
+	stdOut := testutil.WithoutBrokerReferences(kafkaCtl.GetStdOut())
+
+	expected := `
+name: %s
+partitions:
+- id: 0
+  oldestOffset: 0
+  newestOffset: 0
+  leader: any-broker
+  replicas: [any-broker-id]
+  inSyncReplicas: [any-broker-id]
+- id: 1
+  oldestOffset: 0
+  newestOffset: 0
+  leader: any-broker
+  replicas: [any-broker-id]
+  inSyncReplicas: [any-broker-id]
+- id: 2
+  oldestOffset: 0
+  newestOffset: 0
+  leader: any-broker
+  replicas: [any-broker-id]
+  inSyncReplicas: [any-broker-id]
+configs:
+- name: cleanup.policy
+  value: compact
+- name: max.message.bytes
+  value: "10485880"
+- name: min.cleanable.dirty.ratio
+  value: "1.0E-4"
+- name: delete.retention.ms
+  value: "0"
+- name: segment.ms
+  value: "100"`
+
+	testutil.AssertEquals(t, fmt.Sprintf(expected, topicName), stdOut)
+}
+
 func describeTopic(t *testing.T, kafkaCtl testutil.KafkaCtlTestCommand, topicName string) {
 	describeTopic := func(_ uint) error {
 		_, err := kafkaCtl.Execute("describe", "topic", topicName, "-o", "yaml")

internal/topic/topic-operation.go

@@ -1,7 +1,10 @@
 package topic
 
 import (
+	"encoding/json"
 	"fmt"
+	"os"
+	"path"
 	"sort"
 	"strconv"
 	"strings"
@@ -53,9 +56,26 @@ type CreateTopicFlags struct {
 	Partitions        int32
 	ReplicationFactor int16
 	ValidateOnly      bool
+	File              string
 	Configs           []string
 }
 
+type CreateTopicConfig struct {
+	Name       string `json:"Name"`
+	Partitions []struct {
+		ID             int    `json:"ID"`
+		OldestOffset   int    `json:"oldestOffset"`
+		NewestOffset   int    `json:"newestOffset"`
+		Leader         string `json:"Leader"`
+		Replicas       []int  `json:"Replicas"`
+		InSyncReplicas []int  `json:"inSyncReplicas"`
+	} `json:"Partitions"`
+	Configs []struct {
+		Name  string `json:"Name"`
+		Value string `json:"Value"`
+	} `json:"Configs"`
+}
+
 type AlterTopicFlags struct {
 	Partitions        int32
 	ReplicationFactor int16
@@ -111,6 +131,34 @@ func (operation *Operation) CreateTopics(topics []string, flags CreateTopicFlags
 		topicDetails.ConfigEntries[configParts[0]] = &configParts[1]
 	}
 
+	if flags.File != "" {
+		fileContent, err := os.ReadFile(flags.File)
+		if err != nil {
+			return errors.Wrap(err, "could not read topic description file")
+		}
+
+		createTopicConfig := CreateTopicConfig{}
+		ext := path.Ext(flags.File)
+		var unmarshalErr error
+		switch ext {
+		case ".yml", ".yaml":
+			unmarshalErr = yaml.Unmarshal(fileContent, &createTopicConfig)
+		case ".json":
+			unmarshalErr = json.Unmarshal(fileContent, &createTopicConfig)
+		default:
+			return errors.Wrapf(err, "unsupported file format '%s'", ext)
+		}
+		if unmarshalErr != nil {
+			return errors.Wrap(err, "could not umarshal config file")
+		}
+
+		topicDetails.NumPartitions = int32(len(createTopicConfig.Partitions))
+		topicDetails.ReplicationFactor = int16(len(createTopicConfig.Partitions[0].Replicas))
+		for _, v := range createTopicConfig.Configs {
+			topicDetails.ConfigEntries[v.Name] = &v.Value
+		}
+	}
+
 	for _, topic := range topics {
 		if err = admin.CreateTopic(topic, &topicDetails, flags.ValidateOnly); err != nil {
 			return errors.Wrap(err, "failed to create topic")