英文:
security.protocol error when setting up basic Kafka consumer and producer in Go?
问题
我正在尝试在Go中设置一个基本的Kafka客户端，按照这里详细说明的示例进行操作：https://docs.confluent.io/clients-confluent-kafka-go/current/overview.html#go-example-code 和 https://github.com/confluentinc/confluent-kafka-go。
我按照给出的方式编写了消费者和生产者示例，如下所示：
func Produce() {p, err := kafka.NewProducer(&kafka.ConfigMap{"bootstrap.servers": "my-broker-name"})if err != nil {panic(err)}defer p.Close()go func() {for e := range p.Events() {switch ev := e.(type) {case *kafka.Message:if ev.TopicPartition.Error != nil {fmt.Printf("Delivery failed: %v\n", ev.TopicPartition)} else {fmt.Printf("Delivered message to %v\n", ev.TopicPartition)}}}}()topic := "myTopic"for _, word := range []string{"Welcome", "to", "the", "Confluent", "Kafka", "Golang", "client"} {p.Produce(&kafka.Message{TopicPartition: kafka.TopicPartition{Topic: &topic, Partition: kafka.PartitionAny},Value: []byte(word),}, nil)}p.Flush(15 * 1000)}func Consume() {c, err := kafka.NewConsumer(&kafka.ConfigMap{"bootstrap.servers": "my-broker-name","group.id": "myGroup","auto.offset.reset": "earliest",})if err != nil {panic(err)}c.SubscribeTopics([]string{"myTopic", "^aRegex.*[Tt]opic"}, nil)for {msg, err := c.ReadMessage(-1)if err == nil {fmt.Printf("Message on %s: %s\n", msg.TopicPartition, string(msg.Value))} else {fmt.Printf("Consumer error: %v (%v)\n", err, msg)}}c.Close()}
（my-broker-name是我的主机名+端口的替代，我不想在这里包含）
然而，当运行produce函数时，它返回一个错误，说：
Disconnected while requesting ApiVersion: might be caused by incorrect security.protocol configuration (connecting to a SSL listener?) or broker version is < 0.10 (see api.version.request) (after 31ms in state APIVERSION_QUERY)
当运行consume函数时，我收到相同的错误，但还有一些其他的东西，说：
Consumer error: 1/1 brokers are down (<nil>)
我确定这不是真的。
不幸的是，我找不到任何关于这些错误的文档，也不知道如何解决它们。我该如何解决这个错误，以便能够向我的Broker生产和消费消息？
更新：
我获取了我的证书并将其转换为.pem文件，并将ConfigMap更改为以下内容：
p, err := kafka.NewProducer(&kafka.ConfigMap{"bootstrap.servers": "my-broker:32500","security.protocol": "SSL","ssl.certificate.location": "mycert.pem","ssl.ca.location": "ca-chain.pem"})if err != nil {panic(err)}
然而，现在返回的是：
client SSL authentication might be required (see ssl.key.location and ssl.certificate.location and consult the broker logs for more information)
这是否意味着证书有问题？还是我在某个地方漏掉了一步？
英文:
I am attempting to set up a basic Kafka client in Go - following the examples detailed here https://docs.confluent.io/clients-confluent-kafka-go/current/overview.html#go-example-code and https://github.com/confluentinc/confluent-kafka-go.
I wrote the consumer and producer examples the same way they were given, like so
func Produce() {p, err := kafka.NewProducer(&kafka.ConfigMap{"bootstrap.servers": "my-broker-name"})if err != nil {panic(err)}defer p.Close()go func() {for e := range p.Events() {switch ev := e.(type) {case *kafka.Message:if ev.TopicPartition.Error != nil {fmt.Printf("Delivery failed: %v\n", ev.TopicPartition)} else {fmt.Printf("Delivered message to %v\n", ev.TopicPartition)}}}}()topic := "myTopic"for _, word := range []string{"Welcome", "to", "the", "Confluent", "Kafka", "Golang", "client"} {p.Produce(&kafka.Message{TopicPartition: kafka.TopicPartition{Topic: &topic, Partition: kafka.PartitionAny},Value: []byte(word),}, nil)}p.Flush(15 * 1000)}func Consume() {c, err := kafka.NewConsumer(&kafka.ConfigMap{"bootstrap.servers": "my-broker-name","group.id": "myGroup","auto.offset.reset": "earliest",})if err != nil {panic(err)}c.SubscribeTopics([]string{"myTopic", "^aRegex.*[Tt]opic"}, nil)for {msg, err := c.ReadMessage(-1)if err == nil {fmt.Printf("Message on %s: %s\n", msg.TopicPartition, string(msg.Value))} else {fmt.Printf("Consumer error: %v (%v)\n", err, msg)}}c.Close()}
(my-broker-name is a substitute for my hostname + port, which I didn't want to include here)
However when running the produce function it returns an error saying
Disconnected while requesting ApiVersion: might be caused by incorrect security.protocol configuration (connecting to a SSL listener?) or broker version is < 0.10 (see api.version.request) (after 31ms in state APIVERSION_QUERY)
and when running the consume function I receive the same error, but also something that says
Consumer error: 1/1 brokers are down (<nil>)
which I am certain is not the case.
I'm unfortunately unable to find any documentation on what these errors mean, or how to approach fixing them. How do I resolve the error so that I'm able to produce and consume to my Broker?
UPDATE:
I obtained my certificate and converted it to a .pem file, and changed the ConfigMap to the following:
p, err := kafka.NewProducer(&kafka.ConfigMap{"bootstrap.servers": "my-broker:32500","security.protocol": "SSL","ssl.certificate.location": "mycert.pem","ssl.ca.location": "ca-chain.pem"})if err != nil {panic(err)}
However, this is now returning
client SSL authentication might be required (see ssl.key.location and ssl.certificate.location and consult the broker logs for more information)
Does this mean that there is a problem with the Certificate? Or is there a step that I am missing somewhere?
答案1
得分: 1
您需要提供主机名和端口作为引导服务器
"bootstrap.servers": "host1:9092"
要连接到Kafka中的安全端口，您需要提供包含CA文件的信任存储配置，或者对于任何应用程序来说，都需要提供用于安全连接的配置。
https://github.com/FluuxIO/kafka/blob/master/examples/base-client/base-client.go#L6
kafka.ConfigMap{
"bootstrap.servers": "..",
"security.protocol": "SSL",
// 如果您使用SSL身份验证，请在此处提供客户端的密钥
"ssl.key.location": "path-to-private-key.pem",
"ssl.certificate.location": "path-to-public-key.pem",
"ssl.key.password": "如果有的话..",
}
对于您的新错误，请查看以下链接
英文:
You need to provide hostname and port as your bootstrap servers
"bootstrap.servers": "host1:9092"
To connect to secured port in kafka you need to provide truststore configuration that contains your ca file, or any application for secured connection for that matter
https://github.com/FluuxIO/kafka/blob/master/examples/base-client/base-client.go#L6
kafka.ConfigMap{"bootstrap.servers"̇: "..","security.protocol": "SSL",// If you're using SSL authentication, provide the client's key here"ssl.key.location": "path-to-private-key.pem","ssl.certificate.location": "path-to-public-key.pem","ssl.key.password": "if any..",}
For you new error look there
答案2
得分: 0
这里的解决方案是我缺少了ssl.key.location。我不得不向管理员请求密钥。一旦我包含了密钥，一切都正常工作了。我最终的配置如下所示：
c, err := kafka.NewConsumer(&kafka.ConfigMap{"bootstrap.servers": "hostname:port-number","security.protocol": "SSL","ssl.ca.location": "ca-chain.pem","ssl.key.location": "key-location","ssl.certificate.location": "mycert.pem",})if err != nil {panic(err)}
英文:
The solution here was that I was missing ssl.key.location. I had to ask my administrator for the key. Once I included the key everything worked. The final configuration I had looked like the following:
c, err := kafka.NewConsumer(&kafka.ConfigMap{"bootstrap.servers": "hostname:port-number","security.protocol": "SSL","ssl.ca.location": "ca-chain.pem","ssl.key.location": "key-location","ssl.certificate.location": "mycert.pem"})if err != nil {panic(err)}
通过集体智慧和协作来改善编程学习和解决问题的方式。致力于成为全球开发者共同参与的知识库，让每个人都能够通过互相帮助和分享经验来进步。
评论