如何处理Kafka Connect Sink中的背压？

Question

我们构建了一个自定义Kafka Connect接收器,后者又调用远程REST API.如何将背压传播到Kafka Connect基础架构,因此在远程系统比内部消费者向put()传递消息的速度慢的情况下,调用put()的频率较低？Kafka连接文档说我们不应该在put()中阻塞,而是在flush()中阻塞.但是put()中没有阻塞意味着我们必须缓冲数据,如果put()被调用的频率高于flush(),那么在某些时候肯定会导致OOM异常.我已经看到允许kafka使用者在循环()中调用pause()或阻塞.是否有可能在卡夫卡连接接收器中利用它？

Answer 1

我已经看到kafka消费者被允许在loop()中调用pause()或block。是否可以在 kafka 连接接收器中利用它？

原始消费者没有暴露，所以不会。您可以调用/pause整个连接器，尽管我不确定此时未刷新的消息会发生什么。

但是在 put() 中不阻塞意味着我们必须缓冲数据，这肯定会在某些时候导致 OOM 异常

当然可以，但这实际上是保留数据超过必要时间的唯一可行选择。例如，这就是 S3 和 HDFS 连接器的工作原理。

rotate.interval.ms
调用文件提交的时间间隔（以毫秒为单位）...

您的 HTTP 客户端连接很可能会阻塞以发出请求，不是吗？

另一种方法是让您的 HTTP 服务器嵌入 Kafka 消费者，这样它就可以自行轮询消息并在本地对其进行操作，而不需要通过 HTTP 发送请求。