go - 使用 sarama 编写 Kafka 生产者时时间戳无效
问题描述
我有一个正在运行的 Kafka 实例(在本地,在 Docker 中),我使用sarama 包在 Go 中创建了一个生产者。
由于我想在我的主题上使用 Kafka Streams,生产者必须在消息中嵌入时间戳,否则我会收到以下丑陋的错误消息:
org.apache.kafka.streams.errors.StreamsException: 输入记录 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}) 具有无效(负)时间戳。可能是因为使用了 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
我究竟做错了什么?谢谢你的帮助。
解决方案
Sarama 默认为 Kafka 版本 0.8.2。这意味着它在与经纪人交谈时将使用旧的 0.8.2 格式请求。
由于仅在 0.10 中添加了时间戳支持,如果您没有明确指定版本 >= 0.10,您的时间戳将不会转发给代理。
您需要添加config.Version = sarama.V0_10_0_0
到您的代码和时间戳将工作。
推荐阅读
- discord.js - Discord.js 在服务器上的每个通道中创建 webhook
- facebook - 通过 API 创建的直播流的 Facebook 隐私设置
- stm32 - STM32F4 用中断从 uart 接收数据的问题
- apache-kafka - 从 StateStore 更改日志主题具体化 GlobalKTable 并使用exact_once 处理时-其他主题消费者停止工作
- java - Mapbox Android中空对象引用上的NavigationView
- java - SSL peer 错误关闭 - 阿里巴巴 OSS java SDK
- python - 逻辑与,与带有浮点数的 numpy nan 进行比较
- python - Django Rest 框架中的 OTP 验证
- php - 从 laravel 中的选项选择中获取名称和值
- javascript - 使用nodeJS创建依赖下拉菜单过滤器