提问者:小点点

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


我有一个正在运行的Kafka实例(本地,在Docker中),并且我使用sarama包在Go中创建了一个生产者。

当我想在我的主题上使用Kafka Streams时,生产者必须在消息中嵌入时间戳,否则我会收到这个丑陋的错误消息:

org. apache.kafka.stream.error.StreamsException:输入记录消费者记录(topic=crawler_events,分区=0,偏移=0,CreateTime=-1,序列化键大小=-1,序列化值大小=187,header ers=RecordHeaders(header ers = [], 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

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


共1个答案

匿名用户

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

由于仅在0.10中添加了时间戳支持,如果您不指定版本

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

相关问题