inital commit

topic/main.go

@@ -0,0 +1,66 @@
+package main
+
+import (
+	"context"
+	"flag"
+	"log"
+	"net"
+	"strconv"
+
+	"github.com/segmentio/kafka-go"
+)
+
+var (
+	command    = flag.String("c", "", "create or delete")
+	brokers    = flag.String("b", "127.0.0.1:29092", "Brokers list")
+	name       = flag.String("n", "demo", "topic name")
+	partitions = flag.Int("p", 1, "paritions numbers")
+	replicas   = flag.Int("r", 1, "replicas numbers")
+)
+
+func main() {
+	flag.Parse()
+
+	ctx := context.Background()
+
+	conn, err := kafka.DialContext(ctx, "tcp", *brokers)
+	if err != nil {
+		log.Fatal(err)
+	}
+	defer conn.Close()
+
+	controller, err := conn.Controller()
+	if err != nil {
+		panic(err.Error())
+	}
+	var controllerConn *kafka.Conn
+	controllerConn, err = kafka.Dial("tcp", net.JoinHostPort(controller.Host, strconv.Itoa(controller.Port)))
+	if err != nil {
+		panic(err.Error())
+	}
+	defer controllerConn.Close()
+
+	switch *command {
+	case "create":
+		err := controllerConn.CreateTopics(kafka.TopicConfig{
+			Topic:             *name,
+			NumPartitions:     *partitions,
+			ReplicationFactor: *replicas,
+			ConfigEntries: []kafka.ConfigEntry{
+				// {ConfigName: "min.insync.replicas", ConfigValue: "2"},
+				{ConfigName: "segment.bytes", ConfigValue: "2097152"},
+				// {ConfigName: "retention.bytes", ConfigValue: "3145728"},
+			},
+		})
+		if err != nil {
+			log.Fatal(err)
+		}
+	case "delete":
+		err := controllerConn.DeleteTopics(*name)
+		if err != nil {
+			log.Fatal(err)
+		}
+	default:
+		log.Fatal("command must be create or delete")
+	}
+}