Kafka Connect SourceTask的轮询间隔

Question

我正在使用Kafka-Connect API实现自定义源连接器，可用于轮询REST-API并将JSON响应接收到Kafka主题中。现在，我想知道如何实现SourceTask的轮询间隔，JDBC连接器如何提供轮询间隔。我必须在某个地方将线程设置为睡眠状态，但是我必须在哪里执行此操作？

Answer 1

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