Kafka使用者正在读取重新启动时上次提交的偏移量（Java）
Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
offsets.put(new TopicPartition("sometopic", partition), new OffsetAndMetadata(offset));
kafkaconsumer.commitSync(offsets);
我尝试将auto.offset.reset设置为最早和最晚，但这不会更改行为。
我在消费者配置中遗漏了什么吗？
config.put(ConsumerConfig.CLIENT_ID_CONFIG, "CLIENT_ID");
config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
config.put(ConsumerConfig.GROUP_ID_CONFIG, "GROUP_ID");
config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());
config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,CustomDeserializer.class.getName());
config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
共有1个答案
如果要使用commitsync(offset)，则必须小心并阅读其javadoc：
提交的偏移量应该是应用程序将要使用的下一个消息，即lastProcessedMessageOffset+1。
如果不将+1添加到偏移量中，那么在下一次重新启动时，使用者将再次使用上一条消息。正如在另一个答案中提到的，如果您使用commitsync()而不使用任何参数，您就不必担心这一点
-
我有一个用户轮询从订阅的主题。它消耗每条消息并进行一些处理（在几秒内），推送到不同的主题并提交偏移量。 总共有5000条信息, 重新启动前-消耗2900条消息和提交的偏移量 kafka版本（strimzi）>2.0.0 kafka-python==2.0.1
-
我有： 连接的Kafka消费者 此外，我有一个方法，它接受两个参数：消费者和一个重新平衡侦听器，该侦听器跟踪分配给消费者的分区 此方法在计时器上运行，其目标是处理记录，直到没有剩余的记录可读取，或者直到所有分区中的某个最长时间。 由于重新平衡可能发生在使用过程中（在consumer.poll（）已触发多次之后），因此我希望检测此情况，重置并从所有分配的分区（即使已分配）的最后提交偏移量开始重新启动
-
但是如果我们重新启动kafka服务器，使用者会重新读取已经提交的偏移量吗？或者这个选项在这样的情况下工作--服务器重新启动后，只会消耗未读的消息？
-
我使用的是0.10.1.1 API的高级使用者。 奇怪的是，当我关闭应用程序并重新启动它时，偏移量比上次提交的偏移量大一点，我找不到原因。 我在代码中只有一个提交点。 一个分区的示例： 关机前偏移量：3107169023 分区分配时的偏移量：3107180350
-
我正在使用事务性KafkaProducer向主题发送消息。这个很管用。我使用的是具有read_committed隔离级别的KafkaConsumer，而我的seek和seekToEnd方法存在问题。根据文档，seek和seekToEnd方法给出了LSO（上次稳定偏移量）。但这有点让人摸不着头脑。因为它给我的价值总是一样的，主题结束了。无论最后一个条目是（由生产者提交的）还是中止的事务的一部分。例如