Как создать Kafka topi c с помощью Segmentio kafka- go?

Question

Могу ли я получить пример создания топи c с использованием segmentio kafka- go?

Я попытался создать топи c, как показано ниже:

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, я получу эту ошибку:

Not Controller: это неправильный контроллер для этого кластера *

Как правильно передавать адрес кластера для создания топи c?

Kafka Segmentio: 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