Confluent Kafka Golang Client Producer "Брокер: недостаточно синхронизированных реплик" - PullRequest
Новая
       57

Confluent Kafka Golang Client Producer "Брокер: недостаточно синхронизированных реплик"

0 голосов
/ 07 июня 2019

Я пытаюсь протестировать продюсера, пишущего сообщения в тему на кластере kafka с помощью клиента Golang.Это прекрасно работает, когда я пишу тему в локальном кластере, я просто скопировал и вставил пример кода из их 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)
}

Я получаю сообщение на консоли-потребителя noвопросов.

Затем я пытаюсь сделать то же самое, просто используя тему моего удаленного кластера 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

Последнее, что я сделал, это проверилчтобы увидеть, сколько было реплик In-Sync для этой темы.Если я правильно читаю, мин должно быть 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

Есть идеи, на что еще я могу посмотреть?

1 Ответ

1 голос
/ 08 июня 2019

У вас есть min.insync.replicas=2, но в теме есть только одна реплика.

Если у вас есть request.required.acks=all (по умолчанию), то произойдет сбой производителя, потому что он не сможет скопировать то, что вы произвели, ведущему брокеру в минимальный набор требуемых реплик

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

Я считаю, что производитель консоли устанавливает это свойство только на 1

есть 3

На самом деле есть только один. Это идентификатор брокера 3. Вы бы увидели там три отдельных числа в виде ISR, если бы на самом деле было три реплики

...