Golang segmentio/kafka-go 消费者不工作

Question

我正在使用segmentio/kafka-go连接到Kafka。

// to produce messages
topic := "my-topic"
partition := 0

conn, _ := kafka.DialLeader(context.Background(), "tcp", "localhost:9092", topic, partition)

conn.SetWriteDeadline(time.Now().Add(10*time.Second))
conn.WriteMessages(
    kafka.Message{Value: []byte("one!")},
    kafka.Message{Value: []byte("two!")},
    kafka.Message{Value: []byte("three!")},
)

conn.Close()

我可以使用此代码生成到我的 Kafka 服务器中。

// to consume messages
topic := "my-topic"
partition := 0

conn, _ := kafka.DialLeader(context.Background(), "tcp", "localhost:9092", topic, partition)

conn.SetReadDeadline(time.Now().Add(10*time.Second))
batch := conn.ReadBatch(10e3, 1e6) // fetch 10KB min, 1MB max

b := make([]byte, 10e3) // 10KB max per message
for {
    _, err := batch.Read(b)
    if err != nil {
        // err -> "invalid codec"
        break
    }
    fmt.Println(string(b))
}

batch.Close()
conn.Close()

但我无法使用上面的代码进行消费。我收到错误invalid codec。可能是什么原因？

如果相关的话，我将最小批量大小调整为 1，以便它尝试消耗一些东西。

Answer 1

只是猜测：尝试添加导入来加载压缩编解码器，以防您的主题使用压缩。

导入_“github.com/segmentio/kafka-go/snappy”