Confluent Kafka Golang 客户端生产者 "Broker: Not enough in-sync replicas"
Confluent Kafka Golang Client Producer "Broker: Not enough in-sync replicas"
我正在尝试测试生产者使用 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
(这是默认设置),那么生产者将失败,因为它无法将您向领导代理生产的内容复制到所需副本的最小集合
https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md#topic-configuration-properties
我相信控制台制作人只会将 属性 设置为 1
there are 3
实际上只有一个。那是经纪人 ID 3。如果实际上有三个副本,你会看到总共三个单独的数字作为 ISR
或者,如果您使用的是 AWS 的 MSK,当每个代理的 EBS 存储完全用于其中一个代理时,可能会出现这种情况,可能的解决方法是增加其存储。
我正在尝试测试生产者使用 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
(这是默认设置),那么生产者将失败,因为它无法将您向领导代理生产的内容复制到所需副本的最小集合
https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md#topic-configuration-properties
我相信控制台制作人只会将 属性 设置为 1
there are 3
实际上只有一个。那是经纪人 ID 3。如果实际上有三个副本,你会看到总共三个单独的数字作为 ISR
或者,如果您使用的是 AWS 的 MSK,当每个代理的 EBS 存储完全用于其中一个代理时,可能会出现这种情况,可能的解决方法是增加其存储。