用sarama编写Kafka制片人时的无效时间戳

Question

我有一个Kafka实例正在运行(在本地，在一个Docker中)，我使用萨拉马包在Go中创建了一个生产者。

由于我想在我的主题中使用Kafka流，所以生产者必须在消息中嵌入一个时间戳，否则我会得到一条丑陋的错误消息：

org.apache.kafka.streams.errors.StreamsException:输入记录ConsumerRecord(主题= crawler_events，分区= 0，偏移量= 0，CreateTime = -1，序列化键大小= -1，序列化值大小= 187，headers = RecordHeaders(headers = []，isReadOnly = false)，key = null，value = {XXX})具有无效(负)时间戳。可能是因为在没有嵌入时间戳的情况下，使用了预0.10生产者客户端将此记录写入Kafka，或者是因为输入主题是在将Kafka集群升级到0.10+之前创建的。使用不同的TimestampExtractor来处理这些数据。

下面是在我的围棋程序中发送消息的部分代码：

// Init a connection to the Kafka host,
// create the producer,
// and count successes and errors in delivery
func (c *kafkaClient) init() {
    config := sarama.NewConfig()
    config.Producer.Return.Successes = true
    c.config = *config
    var err error
    c.producer, err = sarama.NewAsyncProducer(c.hosts, &c.config)
    if err != nil {
        panic(err)
    }
    go func() {
        for range c.producer.Successes() {
            c.successes++
        }
    }()
    go func() {
        for range c.producer.Errors() {
            c.errors++
        }
    }()
}

// Send a message to the Kafka topic, WITH TIMESTAMP
func (c *kafkaClient) send(event string) {
    message := &sarama.ProducerMessage{
        Topic: c.topic,
        Value: sarama.StringEncoder(event),
        Timestamp: time.Now(),
    }
    c.producer.Input() <- message
    c.enqueued++
}

如您所见，我试图发送的时间戳是time.Now()。

当我运行控制台使用者来查看接收到的时间戳时：

docker-compose exec kafka /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 --topic crawler_events \
  --from-beginning --property print.timestamp=true

我看他们都是"-1"：

CreateTime:-1   {"XXX"}

当使用控制台生成器向主题添加消息时，我有预期的时间戳，如下所示：

CreateTime:1539010180284        hello

我做错了什么？谢谢你的帮助。

Answer 1

Sarama默认为Kafka版本0.8.2。这意味着在与代理交谈时，它将使用旧的0.8.2格式的请求。

由于时间戳支持只在0.10中添加，如果不显式地指定>= 0.10版本，则不会将时间戳转发给代理。

您需要将config.Version = sarama.V0_10_0_0添加到代码中，时间戳就可以工作了。