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

Question

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.

Answer 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.