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

I have a Kafka instance running (locally, in a Docker) and I created a producer in Go, using the sarama package.

As I want to use Kafka Streams on my topic, the producer has to embed a timestamp in the messages, or I get this ugly error message:

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.

Here is the portion of code sending the message in my Go program:

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

As you can see, the timestamp I try to send is time.Now().

When I run the console consumer to see the received timestamps:

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

I see they are all "-1":

CreateTime:-1   {"XXX"}

When adding a message to the topic with the console producer, I have the expected timestamps like:

CreateTime:1539010180284        hello

What am I doing wrong? Thanks for your help.

1个回答

Sarama defaults to Kafka Version 0.8.2. It means it will use the old 0.8.2 format requests when talking to brokers.

As timestamp support was only added in 0.10, if you don't specify a Version >= 0.10 explicitly, your timestamp won't be forwarded to the brokers.

You need to add config.Version = sarama.V0_10_0_0 to your code and timestamps will work.

doulue1949
doulue1949 是! 非常感谢。 而且,不需要Timestamp行:time.Now()。 实际上,“时间戳是代理分配给消息的时间戳”(文档)。
接近 2 年之前 回复
Csdn user default icon
上传中...
上传图片
插入图片
抄袭、复制答案,以达到刷声望分或其他目的的行为,在CSDN问答是严格禁止的,一经发现立刻封号。是时候展现真正的技术了!
立即提问
相关内容推荐