为什么我向 kafka 生成 65536 条消息但只消耗了数百条消息？答案

【问题标题】：why I produces 65536 messages to kafka but only got hundreds consumed?为什么我向 kafka 生成 65536 条消息但只消耗了数百条消息？
【发布时间】：2018-02-07 15:21:59
【问题描述】：

卡夫卡版本：1.0.0 萨拉玛版本：1.15.0 去版本：1.9.1

代码示例如下：

func main() {
    config := sarama.NewConfig()
    config.Version = sarama.V1_0_0_0
    config.Producer.Return.Successes = true
    config.Producer.RequiredAcks = sarama.WaitForAll
    // config.Producer.Flush.Frequency = 10 * time.Second
    // config.Producer.Flush.Bytes = 1024 * 1024
    // config.Producer.Flush.MaxMessages = 1024
    producer, err := sarama.NewAsyncProducer(strings.Split(*brokers, ","), config)
    if err != nil {
        panic(err)
    }

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

    var (
        wg                          sync.WaitGroup
        enqueued, successes, errors int
    )

    wg.Add(1)
    go func() {
        defer wg.Done()
        for range producer.Successes() {
            successes++
        }
    }()

    wg.Add(1)
    go func() {
        defer wg.Done()
        for err := range producer.Errors() {
            log.Println(err)
            errors++
        }
    }()
    counter := 0

ProducerLoop:
    for {
        if counter >= 65536 {
            producer.AsyncClose() // Trigger a shutdown of the producer.
            break ProducerLoop
        }
        message := &sarama.ProducerMessage{
            Topic: *topics,
            // Key:       sarama.StringEncoder(fmt.Sprintf("%d", counter)),
            // Partition: int32(counter),
            Value: sarama.StringEncoder(fmt.Sprintf("%d,%d", counter, time.Now().UnixNano())),
            // Timestamp: time.Now(),
        }
        select {
        case producer.Input() <- message:
            enqueued++

        case <-signals:
            producer.AsyncClose() // Trigger a shutdown of the producer.
            break ProducerLoop
        }
        if *verbose {
            fmt.Printf(".")
        }
        if *sleep {
            // fmt.Println(100 * time.Millisecond)
            time.Sleep(1 * time.Millisecond)
        }
        counter++
    }

    wg.Wait()

    log.Printf("Successfully produced: %d; errors: %d\n", successes, errors)
}

从日志中我发现有65536条消息发送到kafka，但是当我使用kafka官方消费者客户端消费时，只收到了一百条消息，我很困惑

我使用官方的消费者工具来做到这一点：

$KAFKA_HOME/bin/kafka-console-consumer.sh --bootstrap-server $KAFKA_BROKERS --topic fire-8  --from-beginning

关于生成时的偏移量和分区，我也记录了它们。太长了，贴在这里。计数匹配，我没有发现任何奇怪的东西

感谢任何解释。

我也在 GitHub 上发布了issue。

更新 1

我发现如果我设置睡眠，消耗和生产之间的差距就会缩小。

更新 2

我在日志中找到了我发送的数据，但我无法消费，我猜是消费者出了点问题

更新3

我在较旧的 kafka 集群 (0.10.1.0) 上重放了上述步骤，一切正常

【问题讨论】：

尝试同时记录消息计数和提交的偏移量，这样您就可以确保将这些消息提交给 kafka ，这样您就可以确定它是生产者还是消费者问题
Nullne，你能用消费者代码/命令/配置更新你的问题吗？我想确保问题不在于消费。
@DanielWojda 我用的是官方消费工具，上面更新了
@LijuJohn 谢谢你，我记录了偏移量和分区并在上面更新了

标签： apache-kafka producer

【解决方案1】：

应该是kafka go客户端的bug：sarama。我从我发布的GitHub issue 中找到了解决方案：删除 kafka 版本配置，它可以工作。但这失去了新版本带来的一些新功能。如果没有指定使用的版本，则默认是最旧的

【讨论】：