Kafka消费者在重新启动时跳过消息

Question

我有一个Kafka集群正在运行，当重新启动应用程序（消费者）时，它会跳过一些在应用程序关闭时推送到主题的消息。

当应用程序启动时，我可以看到它读取带有偏移量100的消息，然后将偏移量101推送到__consumer_offsets。然后当应用程序关闭时，带有偏移量101、102和103的消息被推送到主题。重启应用程序后，它读取101并将其偏移量设置为104，因此跳过102和103。

这是我的配置：

config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaCluster);
config.put(ConsumerConfig.GROUP_ID_CONFIG, my-consumer);
config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, deserializer);

Answer 1

查看可用信息，您的消费者似乎没有跳过任何消息。

使用偏移量100的消息后，内部主题__consumer_offsets存储此消费者的偏移量101。偏移量101是消费者将在该主题上读取的下一个偏移量。

重新启动后，并且主题中还有3条消息，消费者开始处理偏移量101，并且随后也应该处理其他消息。但是，根据您的提交策略，它会向内部主题__consumer_offsets报告要读取的下一条消息具有偏移量104。它不会特别声明101、102和103，因为所有消息都被一次轮询。