在我们的项目中,有活动Kafka服务器(PR)和被动Kafka服务器(DR),两个Kafka代理在我们的项目中配置有相同的组名称、主题名称和分区。当从PR切换到DR时,_consumer_offsets会手动设置在DR上。
我的问题是,Kafka消费者是否能够无缝地从上次读取的位置消费消息?
我的问题是,Kafka消费者是否能够无缝地从上次读取的位置消费消息?
在两个集群之间复制消息时,无法确保偏移量保持同步。
例如,如果一个主题在活动集群上存在一段时间,某些分区的日志起始偏移量可能不为0(由于保留策略而删除了一些记录)。因此,在复制此主题时,两个集群之间的偏移量将不相同。当在两个集群之间复制时,消息丢失或重复也可能发生,因为无法实现精确一次语义。
因此,不能只复制__consumer_offsets
主题,这样是行不通的。消费者组位置必须在两个集群之间明确地进行“转换”。虽然可以通过直接提交来“手动”重置它们,但不建议这样做,因为找到新位置并不明显。
相反,您应该使用支持“偏移量转换”的复制工具,以确保消费者可以在两个集群之间无缝切换。
例如,Mirror Maker 2是官方的Kafka集群镜像工具,通过RemoteClusterUtils
支持偏移量转换。您可以在KIP中找到详细信息。就本身而言,依赖于两个集群具有相同的偏移量是有问题的。
偏移量 - 是相对特征。它不是消息的一部分。它实际上是文件内的位置。而这些文件,Kafka日志文件,也会进行轮换和保留。不能保证这些日志文件在任何给定时间点都是相同的。Kafka并不声称解决这样的问题。
此外,从CAP的角度来看,解决起来也很棘手。
除非你想要严格的物理复制,否则这也是没有意义的。
这就是为什么Kafka多集群工具通常涉及逻辑复制的原因。我没有使用过Mirror Maker(MM
),但我使用过Replicator(这是Confluent提供的更高级的商业工具),它有一个名为offset translation的功能,谁能猜到,就像MM
一样。
Replicator执行以下操作:
__consumer_timestamps
主题中读取消费者偏移量和时间戳信息,以了解消费者组的进度。__consumer_offsets
主题,前提是该组中没有任何连接到目标集群的消费者。注意:您需要为Kafka消费者添加一个拦截器。