使用 sarama 编写 Kafka 生产者时时间戳无效

golang学习网今天将给大家带来《使用 sarama 编写 Kafka 生产者时时间戳无效》，感兴趣的朋友请继续看下去吧！以下内容将会涉及到等等知识点，如果你是正在学习Golang或者已经是大佬级别了，都非常欢迎也希望大家都能给我建议评论哈~希望能帮助到大家！

我有一个正在运行的 kafka 实例（在本地，在 docker 中），并且我使用 sarama 包在 go 中创建了一个生产者。

由于我想在我的主题上使用 kafka streams，生产者必须在消息中嵌入时间戳，否则我会收到这个丑陋的错误消息：

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 来处理此数据。

这是在我的 go 程序中发送消息的代码部分：

// 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

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

解决方案

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

由于时间戳支持仅在 0.10 中添加，如果您没有明确指定 >= 0.10 的版本，您的时间戳将不会转发给代理。

您需要将 config.Version = sarama.V0_10_0_0 添加到您的代码中，时间戳才会起作用。

以上就是本文的全部内容了，是否有顺利帮助你解决问题？若是能给你带来学习上的帮助，请大家多多支持golang学习网！更多关于Golang的相关知识，也可关注golang学习网公众号。