1

我可以得到一个使用segmentio的kafka-go创建主题的例子吗？

我尝试创建如下主题：

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

但这只有在给定的host:port是 Kafka Leader 时才有效。如果host:port不是 Kafka Leader，那么我将收到此错误：

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

传递集群地址以创建主题的正确方法是什么？

Kafka Segmentio：github.com/segmentio/kafka-go

4

2 回答 2

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

于 2020-05-06T08:49:31.070 回答
0

就像 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())
}
于 2021-09-23T00:05:57.453 回答