是否有无论如何结合KafkaSource和ThrottledIterator在Flink 1.15?

Question

有很多关于速率限制/限制Kafka消费者的老话题

• 如何在链接上使用速率限制器?
• 动态节流flink kafka源
• 等。

但是它们都不能在案例1.15中使用:

• KafkaFetcher不暴露emitRecord
  • 现在它位于KafkaRecordEmitter
• FlinkKafkaConsumer已弃用
  • KafkaSource是首选方法
• KafkaSource显式地在createReader中创建KafkaRecordEmitter

所以，我的问题是，无论如何都有ThrottledIterator和KafkaSource的组合吗?

Answer 1

您不应该在主处理线程中应用速率限制，因为这会阻塞检查点。相反，您应该在反序列化模式中应用速率限制。

您可以在反序列化模式的open方法中设置速率限制器，并在deserialize方法中获取它。(但我会使用Guava RateLimiter而不是节流迭代器。)