如何解决在Camel RoutePolicy中访问本机KafkaConsumer以更改轮询行为
我使用Camel RoutePolicy“监视”我的骆驼处理管道中的连续失败次数。
达到故障阈值时,我想暂停处理一段配置的时间,因为这可能意味着来自另一个系统的数据尚未准备好,因此每条消息都会失败。
由于渠道的来源是 Kafka主题,所以我不应该只停掉整条路线,因为经纪人会认为我的消费者已经死亡并重新平衡。
“暂停”主题消费的最佳方法似乎是pause [KafkaConsumer][3]
(本机语言,而不是骆驼语言之一)。这样,消费者继续轮询经纪人,但不提取任何消息。正是我所需要的。
但是我可以从RoutePolicy上下文中访问本地[KafkaConsumer][3]
来调用pause
和resume
方法吗?
spring-kafka listener containers expose these methods,也可以从Camel使用它们。
解决方法
尚不支持,必须先将这两种方法添加到camel-kafka使用者。
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。