Неверная отметка времени при написании кафки-продюсера с сарамой - PullRequest
Новая
       40

Неверная отметка времени при написании кафки-продюсера с сарамой

0 голосов
/ 08 октября 2018

У меня запущен экземпляр Kafka (локально, в Docker), и я создал продюсер в Go, используя пакет sarama .

Поскольку я хочу использовать потоки Kafka на моемВ теме, продюсер должен вставить метку времени в сообщения, или я получаю это уродливое сообщение об ошибке:

org.apache.kafka.streams.errors.StreamsException: входная запись ConsumerRecord (topic = crawler_events,раздел = 0, смещение = 0, CreateTime = -1, размер сериализованного ключа = -1, размер сериализованного значения = 187, заголовки = RecordHeaders (headers = [], isReadOnly = false), ключ = null, значение = {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 Ответ

0 голосов
/ 08 октября 2018

Сарама по умолчанию использует Кафку версии 0.8.2.Это означает, что он будет использовать старые запросы формата 0.8.2 при общении с брокерами.

Поскольку поддержка меток времени была добавлена ​​только в 0.10, если вы не укажете явно версию> = 0.10, ваша метка времени не будетпереадресация брокерам

...