无效的时间戳写Kafka生产者与萨拉玛

4ngedf3f  于 2021-06-06  发布在  Kafka
关注(0)|答案(1)|浏览(407)

我有一个kafka示例正在运行(本地,在docker中),我使用sarama包在go中创建了一个producer。
当我想在我的主题中使用kafka流时,生产者必须在消息中嵌入一个时间戳,否则我会得到这个丑陋的错误消息:
org.apache.kafka.streams.errors.streamsexception:输入记录consumerrecord(topic=crawler\u events,partition=0,offset=0,createtime=-1,serialized key size=-1,serialized value size=187,headers=recordheaders(headers=[],isreadonly=false),key=null,value={})具有无效(负)时间戳。可能是因为使用了0.10以前的producer客户机将此记录写入kafka而没有嵌入时间戳,或者是因为输入主题是在将kafka集群升级到0.10+之前创建的。使用不同的时间戳提取器来处理此数据。
以下是在我的围棋程序中发送消息的代码部分:

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

在使用console producer向主题添加消息时,我有如下预期的时间戳:

CreateTime:1539010180284        hello

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

zz2j4svz

zz2j4svz1#

萨拉玛默认为Kafka版本0.8.2。这意味着它在与代理交谈时将使用旧的0.8.2格式请求。
由于时间戳支持仅在0.10中添加,如果未显式指定版本>=0.10,则时间戳不会转发给代理。
你需要加上 config.Version = sarama.V0_10_0_0 你的代码和时间戳将工作。

相关问题