当前位置：首页 > 文章列表 > Golang > Go问答 > Confluence Kafka Golang 客户端生产者“经纪人：没有足够的同步副本”

Confluence Kafka Golang 客户端生产者“经纪人：没有足够的同步副本”

来源：stackoverflow 2024-03-16 14:36:27 0浏览收藏

在使用 Go 语言客户端测试将消息写入 Kafka 集群主题时，开发者遇到了“经纪人：没有足够的同步副本”错误。尽管本地主题的写入成功，但使用远程集群主题时却出现了该错误。检查后发现，尽管主题的最小同步副本数设置为 2，但实际只有一个副本。由于生产者将“request.required.acks”属性设置为默认值“all”，因此在无法将消息复制到所需副本数时导致失败。

问题内容

我正在尝试使用 golang 客户端测试生产者将消息写入 kafka 集群上的主题。这对于本地集群上的主题的写入效果很好，我只是从他们的 github 存储库中复制并粘贴了示例代码。

package main

import (
    "fmt"
    "gopkg.in/confluentinc/confluent-kafka-go.v1/kafka"
)


func main() {

    p, err := kafka.newproducer(&kafka.configmap{"bootstrap.servers":"localhost"})
    if err != nil {
        panic(err)
    }

    defer p.close()

    // delivery report handler for produced messages
    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)
                }
            }
        }
    }()

    // produce messages to topic (asynchronously)
    topic := "test"
    for _, word := range []string{"test message"} {
        p.produce(&kafka.message{
            topicpartition: kafka.topicpartition{topic: &topic, partition: kafka.partitionany},
            value:          []byte(word),
        }, nil)
    }

    // wait for message deliveries before shutting down
    p.flush(15 * 1000)
}

我在控制台上收到消息 - 消费者没有问题。

然后我尝试做同样的事情，只是使用我的远程 kafka 集群主题（注意我也尝试过不使用字符串中的端口）：

p, err := kafka.newproducer(&kafka.configmap{"bootstrap.servers":"hostname.amazonaws.com:9092,hostname2.amazonaws.com:9092,hostname3.amazonaws.com:9092"})

它打印以下错误：

delivery failed: test[0]@end(broker: not enough in-sync replicas)

控制台制作人没有任何问题：

./bin/kafka-console-producer.sh --broker-list hostname.amazonaws.com:9092,hostname2.amazonaws.com:9092,hostname3.amazonaws.com:9092 --topic test
>proving that this works

控制台消费者收到它：

bin/kafka-console-consumer.sh --bootstrap-server hostname.amazonaws.com:9092,hostname2.amazonaws.com:9092,hostname3.amazonaws.com:9092 --topic test --from-beginning 

proving that this works

我做的最后一件事是检查该主题有多少个同步副本。如果我没看错的话，最小值应该是 2，而且有 3 个。

./bin/kafka-topics.sh --describe --bootstrap-server HOSTNAME1.amazonaws.com:9092,HOSTNAME2.amazonaws.com:9092,HOSTNAME3.amazonaws.com:9092 --topic test 
Topic:test      PartitionCount:1        ReplicationFactor:1     Configs:min.insync.replicas=2,flush.ms=10000,segment.bytes=1073741824,retention.ms=86400000,flush.messages=9223372036854775807,max.message.bytes=1000012,min.cleanable.dirty.ratio=0.5,unclean.leader.election.enable=true,retention.bytes=-1,delete.retention.ms=86400000,segment.ms=604800000
        Topic: test     Partition: 0    Leader: 3       Replicas: 3     Isr: 3

还有什么我可以研究的想法吗？