list/create kafka topic in go

topics.go

//go:generate go env -w CGO_ENABLED=0 GOPROXY="https://goproxy.cn|https://goproxy.io|direct"
//go:generate rm go.*
//go:generate go mod init kafka.topic
//go:generate go mod tidy
//go:generate go build -trimpath -buildmode pie -installsuffix netgo -tags "osusergo netgo static_build" -ldflags "-s -w -extldflags '-static'" ${GOFILE}
package main

import (
	"flag"
	"fmt"
	"log"
	"strings"

	"github.com/Shopify/sarama"
)

var (
	brokers  string
	topic    string
	action   string
)

func init() {
	flag.StringVar(&action, "action", "list", "action to do")
	flag.StringVar(&brokers, "brokers", "localhost:9092", "List of brokers to connect,comma separator string")
	flag.StringVar(&topic, "topic", "ttttt", "Topic name")

	flag.Parse()
}

func main() {
	switch action {
	case "list", "l":
		list()
	case "create", "c":
		create()
	default:
		fmt.Println("not support action:", action)
	}
}

func list() {
	config := sarama.NewConfig()
	config.Consumer.Return.Errors = true

	//get broker
	cluster, err := sarama.NewConsumer(strings.Split(brokers, `,`), config)
	if err != nil {
		panic(err)
	}

	defer func() {
		if err := cluster.Close(); err != nil {
			panic(err)
		}
	}()

	//get all topic from cluster
	topics, _ := cluster.Topics()
	for index := range topics {
		fmt.Println(topics[index])
	}
}

func create() {
	config := sarama.NewConfig()
	config.Version = sarama.V2_1_0_0
	admin, err := sarama.NewClusterAdmin(strings.Split(brokers, `,`), config)
	if err != nil {
		log.Fatal("Error while creating cluster admin: ", err.Error())
	}
	defer func() { _ = admin.Close() }()
	err = admin.CreateTopic(topic, &sarama.TopicDetail{
		NumPartitions:     1,
		ReplicationFactor: 1,
	}, false)
	if err != nil {
		log.Fatal("Error while creating topic: ", err.Error())
	}
}