正如问题How to manually control the offset commit with camel-kafka?中所问的，我希望使用camel-kafka手动提交偏移量。问题是聚合器和kafka生产者是在独立的线程中工作的，kafka消费者负责抵消承诺。java.util.ConcurrentModificationException: KafkaConsumer is not safe for multi-threaded access 在聚合和分派提交
浏览 23提问于2019-06-22得票数 0
我正在使用apache camel集成我的kafka消息。此外，我还使用JAVA DSL来使用来自kafka端点的消息。使用apache kafka API，它知道如何提交具有给定属性切换的消费者偏移量。
如果我在camel-kafka组件中取消了自动提交，那么我如何在apcahe camel中提交偏移量呢？我使用下面的端点在Apache Camel中禁用自动<e
浏览 9提问于2015-09-21得票数 5
因此，在某些异常情况下，我们不希望提交偏移量。我已经设置了autoCommitEnable=false，而不是处理异常。但是不管抛出哪个异常，偏移量都将被提交(只有堆栈跟踪会被打印出来)。为了处理这个场景，我修改了camel-kafka中的源代码。在异常情况下，我如何阻止骆驼提交偏移量？
我可以看到春天有财产-kakfa "ackOnError"。我们在骆驼-卡夫卡有类似的财产吗(我找不到)？请帮帮忙。
浏览 0提问于2017-05-02得票数 0
我使用的是camel kafka组件，但我不清楚在提交偏移量时，幕后发生了什么。如下所示，我正在聚合记录，我认为对于我的用例，只有在将记录保存到SFTP之后才能提交偏移量。是否可以手动控制何时可以执行提交？
浏览 49提问于2017-08-30得票数 2
回答已采纳
1回答
“简单”一书中的示例使用HashMap来维护已处理的当前偏移量，并在撤消分区时提交当前状态。我担心的是：
它说要使用commitSync来确保在再平衡进行之前已经提交了补偿。但是，这只在该使用者内部是同步的
浏览 1提问于2018-04-11得票数 15
回答已采纳
但是，我怀疑这是否是正确的处理方式，因为当我提交消息时，我的线程池可能仍然在处理一些消息。
浏览 1提问于2018-08-24得票数 0
我们使用Spring Cloud Stream Kafka Binder (带项目反应堆集成，即Flux streams)和manual offset commits (即autoCommitOffset我们正在尝试编写一个与Embedded Kafka from spring-kafka-test的集成测试，通过手动读取消费者组偏移量，使用admin client，在测试向我们的主题发送消息之前和之后使用awaitility，我们现在等待10秒来轮询偏移量，这似乎绕过了我们的大多数问题，因为偏移
浏览 15提问于2019-07-01得票数 1
回答已采纳
2回答
一种选择是运行两个消费者，但在这种情况下，我需要为它们提供两个轮询线程，因此应该处理多线程。有没有更简单的方法？
浏览 1提问于2021-07-13得票数 0
现在，在处理错误之后，我们可以提交偏移量或忽略偏移量。但我想知道，当我们使用多个线程(ConcurrentKafkaListenerContainerFactory.setConcurrency).时，它将如何表现如果由于不同的原因，最新的记录在前一条记录之前出错，我们提交了最新记录的偏移量，但是前一条记录失败了，出现了致命错误，我们实际上不应该提交任何偏移量。ackOnError结合使用
浏览 44提问于2019-01-02得票数 1
回答已采纳
1回答
卡夫卡春中的延迟ACK
、、、、
我使用Spring和Spring作为批处理服务，从Kafka收集数据，直到满足特定条件，然后转储数据。
当数据离开我的服务时，我想确认提交，但它可能会在内存中停留5-10分钟。仅考虑到分区/偏移量信息，还有其他方法可以确认/提交Spring的偏移量吗？
浏览 0提问于2018-10-31得票数 0
回答已采纳
我们假设有一个使用者发送一个提交偏移量10的请求。如果存在通信问题，而代理没有收到请求，当然也没有响应。之后，我们有另一个消费者处理，另一个批，并成功提交偏移量20。Q:，我想知道是否有一种方法或属性可以处理，这样我们就可以在提交之前检查日志中以前的偏移量是否已提交，在我们的情况下，偏移量是20吗？
浏览 1提问于2020-10-04得票数 2
回答已采纳
好吧，当使用者异步地将长任务委托给参与者或线程时，对于这个主题，即使使用一个分区和一个使用者，也可能需要相同的解决方案。要实现这一点，还需要考虑的是，提交的偏移确实意味着以前偏移量中的所有先前消息只处理一次，但确实完成了。我听说了自定义的'Committer‘服务，例如，如果提交请求出现在偏移量102，那么这个服务实际上确保提交偏移量102是安全的，方法是检查当前偏移量(
浏览 7提问于2022-09-20得票数 0
回答已采纳
考虑使用具有以下属性的KafkaMessageListenerContainer：
使用MessageListener但是，我唯一的问题是，如果带有偏移量的特定消息(例如15 )首先被成功处理，但是带有14的消息仍在处理，那么会发生什么情况。因此，在这种情况下，我的偏移量将被更新为15，即使14尚未处理。如何处理这种情况？
浏览 1提问于2016-08-14得票数 1
请有人帮助理解何时消费者线程将提交偏移量与弹簧卡夫卡批处理侦听器和自动提交设置为真？使用者线程是否会在将批处理中的所有消息作为默认处理后提交偏移量？我知道如果autoCommit是假的，提交将基于AckModes，但是想知道autoCommit何时为真
enable.auto.commit
浏览 0提问于2019-07-22得票数 3
回答已采纳
2回答
基于并行度管理的Kafka流容错
、、、、
KafkaStream将使用这些事件作为终端操作，并将事件集中在一起，比如1000个事件，并将其写入AWS S3。假设应用程序有10个线程都在运行并试图将事件放在S3中，在这种情况下会发生崩溃，因为KafkaStream有( enable.auto.commit = false )，而且我们不能手动提交偏移量，而且所有线程都消耗了来自Kafka主题的消息。在这种情况下，KafkaStream
浏览 4提问于2020-11-23得票数 0
1回答
我有一个应用程序，它有两个线程正在运行。这些线程在它们之间共享一个KafkaConsumer实例。现在，假设主题-分区分配在线程-1中发生了更改。我们如何确保Thread-2能够看到这些更改？也就是说，线程-2的KafkaConsu
浏览 4提问于2021-10-26得票数 2
回答已采纳
我正在尝试验证手动偏移量提交的工作情况。 当我尝试通过使用thread.sleep()/jssc.stop()/在while循环中抛出异常来退出作业时，我看到偏移量正在被提交。spark什么时候实际提交了偏移量？即使批处理因为异常而失败，我也会看到偏移量提交，我希望这里会有一些延迟，因为处理失败了，这里有什么问题吗？
浏览 25提问于2019-10-19得票数 2
回答已采纳
Kafka是否允许一个线程或进程使用来自分区的数据，而另一个线程或进程则负责在数据完全处理后手动提交偏移量？
浏览 0提问于2018-04-17得票数 1
在Python或Java中，您可以获得给定时区的IANA名称的UTC偏移量(目前是这样)(例如，“America/洛杉矶”)。例如，请参见。如何在Ubuntu14.04上使用C/C++进行同样的操作？
编辑：最好是以线程安全的方式(没有环境变量)。
浏览 7提问于2016-11-05得票数 1
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
云直播
对象存储
腾讯会议活动推荐
腾讯云开发者
