Kafka客户端上的哪个API允许将偏移量重置为上次提交的偏移量？

文章/答案/技术大牛

发布

1回答

java、apache-kafka

我使用的是java版本的Kafka客户端版本0.10.0.0。在我的例子中，自动提交是关闭的，我们使用KafKaClient.poll(timout) api读取。但在无法处理消息的情况下，我们希望能够从最后一次提交中重新读取。查看API，我没有发现任何明显的东西。有人能给我指出正确的方向吗？谢谢。

浏览 6提问于2017-01-12得票数 1

回答已采纳

2回答

Apache nifi和kafka微服务

docker、apache-kafka、microservices、apache-nifi

我是Apache Nifi的新手，但我在尝试连接kafka微服务(与生产者)和Apache nifi消费者时遇到了一些问题。/test depends_on: command: python test.py 测试是我的制作人： from在Apache nifi仪表板中，我使用以下参数设置了一个输入: Kafka broker: localhost:9092Group ID: 1 ConsumerKafk

浏览 37提问于2021-07-14得票数 0

1回答

弹簧集成卡夫卡与管理补偿

java、spring、spring-integration、apache-kafka

我正在使用扩展来读取和处理Java应用程序中的Kafka消息。据我所知，它使用了高级的使用者API，它不允许完全管理动物园管理员中的偏移量。在我的例子中，我们有auto.commit.enable=false，以便在消息处理后将偏移量提交给。如果处理失败，那么将不会提交偏移量，我们应该尝试在配置好的时间内再次处理相同的消息，从<

浏览 1提问于2014-06-18得票数 3

3回答

Kafka中的消息处理保证

apache-kafka

在主题类型的消息传递系统中，消费者负责在他连接到主题时报告初始偏移量。如何确保该主题上的所有消息都已被处理？

浏览 1提问于2019-03-14得票数 0

2回答

在Java中手动提交Kafka偏移量

java、apache-kafka、kafka-consumer-api

我有一种情况(不要问)，我有需要从外部服务提交给Kafka的偏移量信息(主题、分区、偏移量)。唯一的方法是创建一个KafkaConsumer对象并调用commitSync / commitAsync吗？或者，有没有一种我错过的机制，我可以在不实例化KafkaConsumer的情况下进行提交？

浏览 0提问于2018-10-31得票数 2

3回答

不清楚卡夫卡中auto.offset.reset和enable.auto.commit的含义

apache-kafka、kafka-consumer-api

我刚接触Kafka，我并不真正理解Kafka配置的含义，有没有人能给我解释一下更容易理解！下面是我的代码： "bootstrap.servers" -> "master:9092,slave1:9092",auto.offset.reset" -> "latest", //earliest or latest "enable.

浏览 0提问于2017-07-05得票数 21

2回答

Kaka auto.offset.reset查询

apache-kafka、kafka-consumer-api

我的项目使用Kafka 0.10.2版本。在消费者中设置enable.auto.commit=false和auto.offset.reset=latest。如果在维护后重新启动使用者，则使用者将再次从第一个偏移量读取，而不是等待最新的偏移消息。为什么会发生这种事？我是否错误地理解了结构？我的要求是，使用者不应该自动提交，应该只读取新的消息放入主题时，它是活动的。

浏览 2提问于2020-04-14得票数 1

回答已采纳

1回答

在Spring中使用registerSeekCallback()进行查找操作

apache-kafka、kafka-consumer-api、spring-kafka

我有N个org.springframework.kafka.listener.ConcurrentMessageListenerContainer和N个MessageListeners在听一个主题的N个分区有时，应用程序需要在与上次提交的偏移量不同的偏移量处启动。我将监听器设置为ConsumerSeekAware，并使用registerSeekCallback()。只有在应用程序启动时才需要重置到不

浏览 2提问于2017-12-06得票数 0

回答已采纳

1回答

卡夫卡消费者-客户端没有在动物园管理员上注册消费者组的偏移量。

apache-kafka、apache-zookeeper、kafka-consumer-api

我试图使用 kafka -clients v.0.10.2.1创建多个不同消费者群体的卡夫卡主题。虽然我无法检索消费者组所做的最后一次偏移量。目前，我的消费财产如下 cproperties.put(ConsumerConfig.GROUP_ID_CONFIGtaskDecoder.getClass()); cproperties.put(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG, &

浏览 0提问于2018-07-17得票数 0

2回答

使用高级API从特定偏移量开始读取kafka消息

apache-kafka

我希望我没有犯错，但我记得在Kafka文档中提到，使用高级API你不能从特定的偏移量开始读取消息，但它提到它会改变。我使用的是kafka 0.8.1.1。提前谢谢。

浏览 0提问于2015-12-16得票数 7

回答已采纳

1回答

手动重置kafka偏移量时，Flink偏移量进入不一致状态

apache-kafka、flink-streaming

由于某些原因，我们不得不从kafka重置命令将kafka偏移重置为最新，因为有大量堆积。我们希望flink应用程序跳过所有这些消息，并从重置后出现的新消息开始。问题是因为flink在内部管理它的偏移量，它不知道这个重置，它现在只从后向读取msg (重置前的偏移点)，现在也不能提交偏移量。因此，每次重启flink应用程序时，它都会

浏览 62提问于2021-06-19得票数 4

回答已采纳

1回答

Kafka consumerGroup丢失了所有分区的已提交偏移量信息，并从一开始就开始使用偏移量。

java、apache-kafka、kafka-consumer-api

我一直在使用kafka主题的事件，在我的应用程序中处理同样的事件已经有一段时间了。该主题有20个分区，我将kafka并发设置为10，因为我使用来自kafka主题的事件和我的应用程序的2个副本。我将提交模式设置为手动立即，因此，一旦应用程序确保事件得到处理，我希望提交分区偏移量。一切都很好，直到卡夫卡服务器<em

浏览 9提问于2021-01-30得票数 2

1回答

了解kakfa结构化流中的检查点设置

apache-spark、spark-streaming-kafka

在这篇()文章中，它说检查点是用来“在我做其他事情之前冻结数据帧的内容”。在spark中有两个不同的检查点概念吗？

浏览 0提问于2019-03-11得票数 0

2回答

Kafka多消费者

apache-kafka、kafka-consumer-api

当我们有多个消费者从单个分区的主题中读取时，所有消费者都有可能得到所有的消息。我已经创建了两个消费者，手动偏移commit.started，第一个消费者，2分钟后启动第二个消费者。第二个消费者正在从第一个消费者停止读取的消息中读取。有没有可能第二个消费者会从头开始阅读所有的信息。我是卡夫卡的新手，请帮帮我。

浏览 24提问于2019-02-09得票数 0

1回答

如何重置卡夫卡偏移量以匹配尾部位置？

java、apache-kafka、apache-storm、apache-zookeeper

我们使用的是暴风与卡夫卡和ZooKeeper。我们遇到了一种情况，我们不得不删除一些主题并用不同的名称重新创建它们。我们的卡夫卡喷口保持不变，除了现在阅读新的主题名称。然而，现在，当尝试从新主题读取时，喷口正在使用来自旧主题分区的偏移量。所以我的主题名分区0的尾部位置是500，但是偏移量大约是10000。是否有方法重置偏移位置，使其与主题的尾部匹配？

浏览 2提问于2016-10-26得票数 4

回答已采纳

1回答

使用kafka-python消费kafka，seek()重置的本地偏移量会提交给kafka吗？

apache-kafka、kafka-python

使用kafka-python消费kafka，seek方法重置的本地偏移量会提交给kafka吗？我正在研究获取双中心机房Kafka集群的rpo索引的解决方案。使用kafka-python获取Kafka集群的最大时间戳，取两个机房的Kafka集群的最大时间戳之差。使用seek()将偏移量<

浏览 8提问于2021-05-20得票数 0

1回答

春天卡夫卡消费者不承诺卡夫卡服务器后，领导人更换

apache-kafka、kafka-consumer-api、kafka-producer-api、spring-kafka

其中有一个包含三个kafka节点的集群：面临着严重的问题，甚至无法在当地复制。事情就是这样发生的： 4)最神秘的是:在应用程序中，一切都好--A工作正常。春季-消费者阅读新的信息并发送给卡夫卡。似乎spring用户将其偏移量保存在内存中，并将其发送给远程kafka</

浏览 0提问于2019-01-24得票数 4

回答已采纳

4回答

卡夫卡消费汽车是如何运作的？

java、apache-kafka、offset、kafka-consumer-api、autocommit

我在读也

浏览 13提问于2017-10-03得票数 28

回答已采纳

1回答

卡夫卡流与消费者群体怪异行为

apache-kafka、kafka-consumer-api、apache-kafka-streams

我有两个高层次的问题被分解成更独立的问题，这两个高级问题都涉及到一个消费群体，一个正在创建和使用。4 myConsumer-7fc71848-465b-4817-93b

浏览 1提问于2017-10-02得票数 0

回答已采纳

1回答

如何在不手动分配分区的情况下精确实现一次Kafka使用者

apache-kafka、kafka-consumer-api

我正在浏览，它解释了如何通过执行以下操作来确保消息被正确地处理一次：如您所见，它显式地指定要从哪个分区读取消息。我觉得这不是个好主意，因为它不允许卡夫卡将公平的分割分配给活跃的消费者。在<em

浏览 12提问于2019-09-20得票数 3

回答已采纳

点击加载更多