go - パーティションから消費できないパッケージ

Question

トピックから消費することができず、コードまたはカフカ構成に問題があるかどうかわかりません。私が得ている問題は、「開始」の印刷ステートメントでスタックしているため、チャネル <-partitionConsumer.Messages() からメッセージを受信して​​いないことです。

これらは、いくつかのメッセージを含む kafka セットアップ ( https://kafka.apache.org/quickstart ) で行った手順です。次のコマンドを実行すると値が表示されるため、それらが存在することは確かです。

bin/kafka-console-consumer.sh --partition 0 --topic test --bootstrap-server localhost:9092 --offset 最も早い

1. bin/zookeeper-server-start.sh config/zookeeper.properties
2. bin/kafka-server-start.sh config/server.properties
3. bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic テスト
4. bin/kafka-console-producer.sh --broker-list localhost:9092 --topic テスト

func RetrieveConsumed() (int){
    consumer, err := sarama.NewConsumer([]string{"localhost:9092"}, nil)
    if err != nil {
        fmt.Println("ERROR")
        panic(err)
    }

    defer func() {
        if err := consumer.Close(); err != nil {
            fmt.Println("ERROR")
        }
    }()

    partitionConsumer, err := consumer.ConsumePartition("test", 0, sarama.OffsetOldest)
    if err != nil {
        fmt.Println("ERROR")
        panic(err)
    }
    defer func() {
        if err := partitionConsumer.Close(); err != nil {
            fmt.Println("ERROR")
        }
    }()

    // Trap SIGINT to trigger a shutdown.
    signals := make(chan os.Signal, 1)
    signal.Notify(signals, os.Interrupt)

    consumed := 0
    ConsumerLoop:
        for {
            fmt.Println("Starts")
            select {
            case msg := <-partitionConsumer.Messages():
                fmt.Printf("Consumed message offset %d\n", msg.Offset)
                consumed++
            case <- partitionConsumer.Errors():
                break ConsumerLoop
            case <-signals:
                break ConsumerLoop
            }
        }

        fmt.Printf("Consumed: %d\n", consumed)
        return consumed
}