我正在尝试测试生产者使用 Golang 客户端向 kafka 集群上的主题写入消息。这可以很好地写入本地集群上的主题，我只是复制并粘贴了他们的示例代码 github repo .

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 

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

请您参考如下方法:

您有 min.insync.replicas=2，但该主题只有一个副本。

如果您有 request.required.acks=all(这是默认设置)，那么生产者将失败，因为它无法将您生产的内容复制到 leader broker 的最小集合所需副本

https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md#topic-configuration-properties

我相信控制台制作人只会将该属性设置为 1

there are 3

实际上只有一个。那是经纪人 ID 3。如果实际上有三个副本，您会在那里看到总共三个单独的数字作为 ISR