如何用西格曼多的卡夫卡围棋创作卡夫卡主题？

Question

我能得到一个例子，用赛段多的卡夫卡-围棋创建一个主题吗？

我试着创建了一个主题如下：

c, _ := kafka.Dial("tcp", "host:port")
kt := kafka.TopicConfig{Topic: "sometopic", NumPartitions: 1, ReplicationFactor: 1}
e := c.CreateTopics(kt)

但这只有在给定的主机:端口是卡夫卡领袖的情况下才能起作用。如果主机:端口不是卡夫卡领袖，那么我将得到以下错误：

：这不是这个集群的正确控制器*

通过集群地址创建主题的正确方法是什么？

卡夫卡·塞格梅戈：github.com/segmentio/kafka-go

Answer 1

这就是你需要的：

func (c *Conn) Controller() (broker Broker, err error)
// Controller requests kafka for the current controller and returns its URL

当您使用Dial在代码中打开连接时，您将随机选择集群中的一个代理。因此，您可能/不可能最终使用实际的Kafka控制器。简单地查找控制器和打开一个新的连接应该是有帮助的。

https://pkg.go.dev/github.com/segmentio/kafka-go?tab=doc#Conn.Controller

Answer 2

就像shmsr说的-你需要得到领导者的连接来创建主题。您可以通过以下方式这样做：

conn, err := kafka.Dial("tcp", "host:port")
if err != nil {
    panic(err.Error())
}
defer conn.Close()

controller, err := conn.Controller()
if err != nil {
    panic(err.Error())
}
controllerConn, err := kafka.Dial("tcp", net.JoinHostPort(controller.Host, strconv.Itoa(controller.Port)))
if err != nil {
    panic(err.Error())
}
defer controllerConn.Close()

topicConfigs := []kafka.TopicConfig{{Topic: "sometopic", NumPartitions: 1, ReplicationFactor: 1}}

err = controllerConn.CreateTopics(topicConfigs...)
if err != nil {
    panic(err.Error())
}