java - Kafka Connect SourceTask 的轮询间隔

Question

我正在使用 Kafka-Connect API 实现自定义源连接器，它可用于轮询 REST-API 并将 JSON 响应发送到 Kafka 主题中。

现在我想知道如何实现 SourceTask 的轮询间隔，JDBC 连接器如何提供轮询间隔。我必须在某处将线程设置为 sleep ，但我必须在哪里执行此操作？

Answer 1

我在我的 SourceTask 实现中解决了这个用例，方法是添加一个 long 类型的私有(private)字段来存储时间戳。在第一次调用 poll() 时，该字段尚未初始化，因此会轮询配置的 REST-API。在第一次调用时，提到的 long 字段会使用当前时间戳进行初始化。在随后的所有 poll() 调用中，将检查上一次调用的这个时间戳。如果自上一个 poll() 以来经过的毫秒数小于两次轮询之间的配置间隔，我会在配置的毫秒数过去后让线程进入 hibernate 状态。