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

Question

我有一个 Kafka 实例运行ning（在本地，在 Docker 中），我使用 sarama package.

在 Go 中创建了一个生产者

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

org.apache.kafka.streams.errors.StreamsException: Input record ConsumerRecord(topic = crawler_events, partition = 0, offset = 0, CreateTime = -1, serialized key size = -1, serialized value size = 187, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = {XXX}) has invalid (negative) timestamp. Possibly because a pre-0.10 producer client was used to write this record to Kafka without embedding a timestamp, or because the input topic was created before upgrading the Kafka cluster to 0.10+. Use a different TimestampExtractor to process this data.

这是我的 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

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

Answer 1

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

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

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

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

Invalid Timestamp when writing a Kafka producer with sarama

go

apache-kafka

apache-kafka-streams

sarama