如何在使用并发时使用ConsumerAwareErrorHandler提交偏移量？

文章/答案/技术大牛

发布

1回答

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

如果出现错误，ConsumerAwareErrorHandler接口将提供对ConsumerRecord和Consumer In handle方法的引用。现在，在处理错误之后，我们可以提交偏移量或忽略偏移量。但我想知道，当我们使用多个线程(ConcurrentKafkaListenerContainerFactory.setConcurrency).时，它将如何表现如果由于不同的原因，最新的记录在前一条记录之前出错，我们提交了最新记录的偏移量，但是前一

浏览 44提问于2019-01-02得票数 1

回答已采纳

2回答

卡夫卡当发生这种情况时，提交的偏移量大于客户端实际处理的最后一条消息的偏移量。

apache-kafka、confluent-platform

我正在阅读权威指南，并发现了这个短语。不知道这是什么时候发生的？我理解了另一种用例，在这种情况下，提交的偏移量较小，在这种情况下，使用者将不得不放弃提交最新的偏移量。

浏览 1提问于2020-06-06得票数 0

回答已采纳

1回答

kafka消费者在重新平衡时获取不同的偏移量

apache-kafka、apache-camel

我有一个微服务应用程序开发使用骆驼-卡夫卡消费卡夫卡的消息。我使用4个并发实例运行此服务，其中每个实例都有1个使用者，这些使用者从具有20个分区的主题中消费。在查看日志时，注意到当应用程序停止重新平衡时，偏移量= 12399，但当应用程序开始消费消息时，偏移量= 12336。currentLeader=LeaderAndEpoch{leader=Optional[testbroker1.local:9093 (id:

浏览 216提问于2020-11-18得票数 0

1回答

使用Akka Stream和Kafka offset提交将事件从Kafka流式传输到Couchbase

scala、apache-kafka、akka、akka-stream、alpakka

我正在尝试设计一个Akka Stream，使用Alpakka来读取kafka主题中的事件，并将它们放到Couchbase中。logging") .runWith(Sink.seq) 我所谓“不知何故”的意思是，这个流实际上是从主题中读取事件，并将它们作为json文档放到Couchbase中，尽管我不知道如何将消费者偏移量提交给如果我已经清楚地理解了隐藏在Kafka消费者偏移量后面的主要思想，在任何故障或重启发生的情况下，流从最后提交的偏移量读取所有消息，并且由于我们

浏览 21提问于2019-10-18得票数 0

1回答

Kafka with Java:如何重新读取数据

java、apache-kafka、commit、kafka-consumer-api

在我的代码中，没有像预期的那样工作，这意味着“如果kafka崩溃，那么从提交的位置重新启动”。但是，当我的DB事务失败时，即使我不使用commitAsync()，指针也会向前移动。<code>A3</code> 但这需要手动维护分区及其偏移量的列表，我猜还有更简单、更优雅的东西？

浏览 31提问于2018-12-21得票数 1

回答已采纳

1回答

重试消费来自Kafka主题的消息

apache-kafka、kafka-consumer-api、spring-cloud-stream

我正在开发一个模块，在这个模块中，它使用来自Kafka主题的消息并发布到下游系统。在下游系统不可用的情况下，消费者不要确认Kakfa消息。因此，当我的消费者在下游系统不可用时收到消息时，kakfa的偏移量将不会被提交。但如果我在下游系统启动后收到新消息，并且当我确认该消息时，最新的偏移量将被提交，消费者将永远不会收到没有偏移量提交的主题中的那些消息。也就是说，假设我的消费者被消耗到偏移量4。当下游

浏览 17提问于2019-08-01得票数 0

2回答

在Kafka消费者中实现并发

parallel-processing、apache-kafka、kafka-consumer-api、consumer

因此，我们尝试使API调用异步，并在其响应中增加偏移量。然而，我们看到了一个问题：通过将API调用设置为异步，我们可以首先获得最后一条记录的响应，并且到那时前一条记录的API调用都还没有启动或完成。如果我们一收到最后一条记录的响应就提交偏移量，那么偏移量就会变成最后一条记录。同时，如果使用者重新启动或分区重新平衡，我们将不会收到提交偏移量为的最后一条记录之前的任何记录。

浏览 2提问于2019-06-19得票数 2

1回答

Kafka -动态/任意分区

asynchronous、apache-kafka、job-scheduling

然而，根据我所读到的，当我真的想要无限或动态的分区时，似乎需要预先定义分区的数量(理想情况下，每条消息都有自己的分区动态创建)。例如，在单个分区上，我可以让使用者读取第一条消息并发出异步请求。它提供了一个回调函数，它将该偏移量提交给Kafka。当该请求等待时，我的使用者将读取下一条消息并发</em

浏览 3提问于2015-05-08得票数 1

回答已采纳

2回答

不提交来自Kafka 10消费者的消费消息

apache-kafka、kafka-consumer-api

如果批处理由于任何原因而失败，我需要再次使用相同的消息集并重复该过程。因此，对于每个批处理，每个分区的from和to偏移量都存储在数据库中。为了实现这一点，我通过将分区分配给读取器来为每个分区创建一个Kafka消费者，基于先前存储的偏移量，消费者将查找到该位置并开始读取。我已经关闭了自动提交，并且我不提交来自消费者的偏移量。对于每个批处理，我为每个分区创建一个新的使用者，从存储的最后一个偏移量读取消息并发布到外部系统。在不提交

浏览 0提问于2016-11-06得票数 3

1回答

如何使用spring KafkaMessageListenerContainer.java确保消息不丢失

kafka-consumer-api、spring-kafka

一旦我从kafka收到一条消息，我需要运行一个长时间运行的进程(最多需要20秒)，只有当这个过程完成时，我才需要将一条消息视为成功。使用MessageListener但是，我唯一的问题是，如果带有偏移量的特定消息(例如15 )首先被成功处理，但是带有14的消息仍在处理，那么会发生什么情况。因此，在这种情况

浏览 1提问于2016-08-14得票数 1

1回答

Kafka有批量消费者吗？

apache-kafka

如果消费者想要处理这些消息并将其提交给其他下游消费者，比如Solr或Elastic-Search，这对他们来说可能是一个很大的问题，因为他们更喜欢批量发送消息，而不是一次只有一个。在内存中对这些消息进行批处理也不是一件容易的事，因为Kafka中的偏移量只有在批处理已经提交时才需要同步，否则具有未提交下游消息的崩溃的kafka消费者(如Solr或ES)的偏移量将已经更新，从而导致消息松散如果消费者在向下游提交消息之后但在更新消息偏移量之

浏览 2提问于2016-03-04得票数 9

1回答

Akka streams Kafka消费者进程并行

apache-kafka、akka、akka-stream、alpakka、akka-kafka

我正在开发一个使用Akka Kafka连接器的Kafka消费者应用程序。我希望消费者并行处理消息。我应该选择哪个消费群体？？如何在消费者端配置并行度？

浏览 16提问于2020-05-06得票数 0

2回答

如何在Spring Kafka中关闭offsets提交，以便在本地存储offsets？

apache-kafka、spring-kafka

我想使用Spring Kafka将事件处理到本地数据库中，并将偏移量存储在数据库本身中，如“在Kafka之外存储偏移量”下的https://kafka.apache.org/25/javadoc/index.htmlSpring Kafka (带有KafkaListener)似乎总是提交偏移量(因此需要group.id)，有没有办法完全禁用提交，允许在没有组的情况下进行侦听？

浏览 46提问于2020-07-03得票数 1

回答已采纳

1回答

从kafka高级消费者获取偏移量

java、apache-kafka

因为我使用的是自定义的提交偏移量属性，所以我想测试一下我的自定义提交偏移量是否工作正常。谁能告诉我如何获得补偿？我遇到了几个kafka工具(如getoffsetshell)，但它对我的测试没有帮助。

浏览 1提问于2015-12-07得票数 1

2回答

我应该如何为卡夫卡原子消费者设定卡夫卡制作者的属性？

apache-kafka

我正在浏览，它解释了如何通过执行以下操作来确保消息被正确地处理一次：通过调用consumer.commitAsync()或consumer.commitSync()手动提交对Kafka的偏移量在此，我将偏移量保存到外部数据库。因

浏览 0提问于2019-10-25得票数 2

1回答

无法从zookeeper检索主题/组/分区的kafka偏移量

apache-zookeeper、apache-kafka

我们正在使用HL消费者，他们会像预期的那样访问kafka队列中的数据，在重启时，他们会从中断的地方重新开始。因此，消费者的行为符合预期。问题是当我们使用zkCli.sh时，我们在zookeeper中看不到偏移量。现在，使用者被设置为仅针对一个主题在一个分区中运行。因为当消费者停止并再次重启时，消费者的行为如预期的那样(即，它从上次运行时停止的偏移量中恢复。我们可以通过查看日志来判断这一点，日志给出了它开始的偏移量，并

浏览 2提问于2015-02-04得票数 1

5回答

使用高级使用者的Kafka延迟队列实现

java、scala、messaging、apache-kafka、kafka-consumer-api

希望使用高级使用者api实现延迟使用者。提交每个偏移量可能会减慢ZK的速度。如果是，我将使用相同的消息两次(

浏览 5提问于2015-08-02得票数 28

回答已采纳

1回答

如何使用Camel-kafka提交消费者偏移量？

java、apache-camel

我正在使用apache camel集成我的kafka消息。此外，我还使用JAVA DSL来使用来自kafka端点的消息。使用apache kafka API，它知道如何提交具有给定属性切换的消费者偏移量。kafka.co

浏览 9提问于2015-09-21得票数 5

1回答

批量记录处理后如何提交kafka偏移量

java、performance、spring-kafka

我正在使用spring-kafka和消费卡夫卡主题的批量记录，并通过AbstractMessageListenerContainer.AckMode.BATCH提交偏移量。在我的例子中，处理批处理记录需要时间(大约20秒)，使用者线程等待批处理过程完成，然后再次执行轮询(在此轮询时提交偏移量)。(在此过程中，使用者线程将一直等待，直到它从ProcessThread获得结果，这会导致性能低下。有没有办法让ProcessThread负责将offset<e

浏览 6提问于2018-10-03得票数 1

1回答

RetryingBatchErrorHandler -偏移量提交处理

spring-kafka

我正在使用spring-kafka 2.3.8，并尝试使用RetryingBatchErrorHandler记录恢复的记录并提交偏移量。您将如何在恢复程序中提交偏移量？

浏览 5提问于2021-10-12得票数 0

点击加载更多