Java 中的 Kafka Connect 动态连接器

Question

这就是开发者指南对动态连接器 https://docs.confluent.io/current/connect/devguide.html#dynamic-connectors 的评价

并非所有连接器都有一组静态分区，因此连接器实现还负责监控外部系统是否有任何可能需要重新配置的更改。例如，在 JDBCSourceConnector 示例中，连接器可能会为每个任务分配一组表。创建新表时，它必须发现这一点，以便它可以通过更新其配置将新表分配给其中一个任务。当它注意到需要重新配置的更改（或任务数量的更改）时，它会通知框架并且框架会更新任何相应的任务

我不确定我应该怎么做。

连接器类（扩展 org.apache.kafka.connect.source.SourceConnector）实现了taskConfigs(int)、start(Map<String,String>)、stop()、config()、version() 和 taskClass()。它没有poll() 或类似方法。

我想在start(Map<String,String>) 中生成一个线程来监视外部系统吗？

如果检测到更改，我该如何“通知框架”。 Java SDK 中是否有一些 API 调用可以执行此操作，或者这是否意味着我必须先调用 stop()，然后再调用 start()？

谢谢。

Answer 1

我想在start(Map<String,String>) 中生成一个线程来监视外部系统吗？

是的，没错。

我如何“通知框架”。

通过 SourceConnector 上下文的requestTaskReconfiguration() 方法。