Kafka版本：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上发布了一个问题。
更新1
我发现，如果我设置睡眠，消耗和生产之间的差距缩小。
更新2
我在日志中找到了我发送的数据，但我无法消费，我猜是消费者出错了

更新3

我在旧的kafka集群（0.10.1.0）上重放上述步骤，一切都完全按照预期工作