如何更改Kafka提交的消费者偏移量与所需偏移量
我有Kafka流应用程序。我的应用程序正在成功处理事件。
如何使用重新处理/跳过事件所需的偏移量更改Kafka committed consumer offset。我试过如何更改topic？的起始偏移量？。但我得到了“节点不存在”错误。请帮帮我。
共有1个答案
您所指的问题/答案基于较旧的Kafka版本。从Kafka 0.9开始，偏移量不提交给ZooKeeper，而是存储在一个名为偏移量主题的特殊Kafka主题中（主题名称为__consumer_offsets）。
自Kafka1.0以来，命令行工具bin/Kafka消费群体。sh有一个新功能，允许设置偏移。查看原始KIP：https://cwiki.apache.org/confluence/display/KAFKA/KIP-122:添加重置消费群体补偿工具
该工具也适用于Kafka0.11（甚至可能是更老的Kafka版本）。
另一种方法是编写自己的工具，使用单个KafkaConsumer和相应的组。id，订阅要修改偏移量的主题，seek（）和commit（）偏移量。（注意，您应该为此使用者禁用自动提交。）
-
我对SpringBoot中的Kafka批处理侦听器有问题。 这是@KafkaListener 对于我的问题，这个解决方案不起作用，因为提交批处理。对于我的解决方案，我需要提交单个消息的偏移量。 我尝试使用
-
null 当侦听器处理记录后返回时提交偏移量。 如果侦听器方法抛出异常，我会认为偏移量不会增加。但是，当我使用下面的code/config/command组合对其进行测试时，情况并非如此。偏移量仍然会得到更新，并且继续处理下一条消息。 我的配置： 验证偏移量的命令： 我使用的是kafka2.12-0.10.2.0和org.springframework.kafka:spring-kafka:1.1
-
我已经将enable.auto.commit设置为true，并将auto.commit.interval.ms设置为10,000（即10秒）。现在我的问题是--消费者是每个记录的提交偏移量，还是根据10秒内消耗的记录数提交并提前偏移量？
-
我们有一个问题，似乎Kafka消费者没有收到发布到某个主题的消息。（我说这是因为我还没有弄清楚这件事的真相，我可能错了。） 我使用Spring for Apache Kafka，而我的消费者实际上是一个用注释的方法。 这个问题是断断续续的，我很难重新创建它。 有没有一种方法让我看看Kafka经纪人的日志，或任何其他工具，以帮助我找出抵消为我的消费者？我想要具体的证据来证明我的消费者是否收到了信息。
-
我正在使用Kafka2.0版和java消费者API来消费来自一个主题的消息。我们使用的是一个单节点Kafka服务器，每个分区有一个使用者。我注意到消费者正在丢失一些消息。场景是：消费者投票主题。我为每个线程创建了一个消费者。获取消息并将其交给处理程序来处理消息。然后使用“至少一次”的Kafka消费者语义来提交Kafka偏移量来提交偏移量。同时，我有另一个消费者使用不同的group-id运行。在这个
-
我正在使用事务性KafkaProducer向主题发送消息。这个很管用。我使用的是具有read_committed隔离级别的KafkaConsumer，而我的seek和seekToEnd方法存在问题。根据文档，seek和seekToEnd方法给出了LSO（上次稳定偏移量）。但这有点让人摸不着头脑。因为它给我的价值总是一样的，主题结束了。无论最后一个条目是（由生产者提交的）还是中止的事务的一部分。例如