我正在尝试使用kafka节点读取来自kafka主题的压缩消息。
问题是,最近插入的消息留在eol之上,并且在插入其他消息之前无法访问。实际上,下线和高位偏移之间存在一个间隙,这会阻止读取最新消息。不清楚为什么会这样。
已使用创建主题
kafka-topics.sh --zookeeper ${KAFKA_HOST}:2181 --create --topic atopic --config "cleanup.policy=compact" --config "delete.retention.ms=100" --config "segment.ms=100" --config "min.cleanable.dirty.ratio=0" --partitions 1 --replication-factor 1
主题中产生了许多键值。有些钥匙是一样的。
var client = new kafka.KafkaClient({kafkaHost: "<host:port>",autoConnect: true})
var producer = new HighLevelProducer(client);
producer.send(payload, function(error, result) {
debug('Sent payload to Kafka: ', payload);
if (error) {
console.error(error);
} else {
res(true)
}
client.close()
});
});
以下是插入的键和值
key - 1
key2 - 1
key3 - 1
key - 2
key2 - 2
key3 - 2
key1 - 3
key - 3
key2 - 3
key3 - 3
然后请求主题键集。
var options = {
id: 'consumer1',
kafkaHost: "<host:port>",
groupId: "consumergroup1",
sessionTimeout: 15000,
protocol: ['roundrobin'],
fromOffset: 'earliest'
};
var consumerGroup = new ConsumerGroup(options, topic);
consumerGroup.on('error', onError);
consumerGroup.on('message', onMessage);
consumerGroup.on('done', function(message) {
consumerGroup.close(true,function(){ });
})
function onError (error) {
console.error(error);
}
function onMessage (message) {)
console.log('%s read msg Topic="%s" Partition=%s Offset=%d HW=%d', this.client.clientId, message.topic, message.partition, message.offset, message.highWaterOffset, message.value);
}
})
结果令人惊讶:
consumer1 read msg Topic="atopic" Partition=0 Offset=4 highWaterOffset=10 Key=key2 value={"name":"key2","url":"2"}
consumer1 read msg Topic="atopic" Partition=0 Offset=5 highWaterOffset=10 Key=key3 value={"name":"key3","url":"2"}
consumer1 read msg Topic="atopic" Partition=0 Offset=6 highWaterOffset=10 Key=key1 value={"name":"key1","url":"3"}
consumer1 read msg Topic="atopic" Partition=0 Offset=7 highWaterOffset=10 Key=key value={"name":"key","url":"3"}
consumer1 read msg Topic="atopic" Partition=0 Offset=0 highWaterOffset=10 Key= value=
consumer1 read msg Topic="atopic" Partition=0 Offset=0 highWaterOffset=10 Key= value=
consumer1 read msg Topic="atopic" Partition=0 Offset=0 highWaterOffset=10 Key= value=
consumer1 read msg Topic="atopic" Partition=0 Offset=0 highWaterOffset=10 Key= value=
有一个代表最新值10的高水位偏移量。然而,消费者看到的偏移值只有7。不知何故,压缩阻止消费者看到最新的消息。
目前还不清楚如何避免这种限制,让消费者看到最新的消息。
谢谢你的建议。谢谢。
2条答案
按热度按时间x33g5p2x1#
不知何故,压缩阻止消费者看到最新的消息。
是的,你错过了一些信息,但你也看到其他人。
压缩是删除先前的关键点。
注意这里没有
url - 1
价值观这是因为您为同一个键发送了新值。
你发了10条信息,所以这个主题的高潮偏移量是10
你的代码看起来不一定是错的,但是你应该有两个以上的3值。打印的偏移量与此逻辑相对应。
一般来说,我建议不要让kafka尝试压缩主题,以每秒10倍的速度编写日志段,也不要使用不同的库,例如
node-rdkafka
oalqel3c2#
在对kafka做了更多的工作之后,kafka节点api似乎有以下行为(我认为这实际上源于kafka本身)。
当在highwateroff之前查询消息时,只有高达highwateroffset的消息才会返回到consumergroup。如果消息没有被复制,这是有意义的,因为组中的另一个使用者不一定会看到这些消息。
仍然可以使用使用者(而不是使用者组)并通过查询特定分区来请求和接收超出highwateroffset的消息。
另外,当偏移量不一定在latestoffset时,“done”事件似乎会被触发。在这种情况下,有必要在message.offset+1处提交进一步的查询。如果你继续这样做,你可以得到所有的消息到最新的偏移量。
我不清楚Kafka为什么会有这种行为,但有一个可能是一些较低层次的细节,表面这种紧急行为。