如何在 Kafka Connect 中动态更新 SinkConnector 的主题？答案

【问题标题】：how to dynamically update topics of SinkConnector in Kafka Connect?如何在 Kafka Connect 中动态更新 SinkConnector 的主题？
【发布时间】：2016-12-31 03:18:09
【问题描述】：

我已经写了一个 Kafka Connect 到消费者主题，但是我的主题会在运行时改变，所以我需要重新配置主题。

我知道使用 RESTful API 可以更新主题，还有其他方法可以实现吗？

【问题讨论】：

标签： apache-kafka apache-kafka-connect

【解决方案1】：

如果您提前知道要切换到的主题集，则可以在连接器配置中指定要使用的主题列表。否则，the REST API 是动态更新配置的唯一方法。

【讨论】：

【解决方案2】：

Kafka Connect 旨在作为服务运行，它还支持用于管理连接器的 REST API。在运行时通过 REST API 进行更新的唯一方法：

PUT /connectors/{name}/config - 在运行时更新连接器的配置参数。

Request Json Object - config(map)

{
    "connector.class": "io.confluent.connect.hdfs.HdfsSinkConnector",
    "tasks.max": "20",
    "topics": "kafkaConnectTopic",
    "hdfs.url": "hdfs://smoketest:9000",
    "hadoop.conf.dir": "/etc/hadoop/conf",
    "hadoop.home": "/etc/hadoop",
    "flush.size": "1000",
    "rotate.interval.ms": "100"
}

Response :

{
    "name": "hdfs-sink-connector",
    "config": {
        "connector.class": "io.confluent.connect.hdfs.HdfsSinkConnector",
        "tasks.max": "20",
        "topics": "kafkaConnectTopic",
        "hdfs.url": "hdfs://smoketest:9000",
        "hadoop.conf.dir": "/etc/hadoop/conf",
        "hadoop.home": "/etc/hadoop",
        "flush.size": "1000",
        "rotate.interval.ms": "100"
    },
    "tasks": [
        { "connector": "hdfs-sink-connector", "task": 1 },
        { "connector": "hdfs-sink-connector", "task": 2 },
        { "connector": "hdfs-sink-connector", "task": 3 }
   ]
}

您可以通过http://docs.confluent.io/3.0.0/connect/userguide.html#connect-administration进一步阅读。

【讨论】：

感谢您的回复，我查看了源代码实际上可以通过修改一点切片代码来实现这一点。但我不确定修改是否会影响其他功能。
您好，如果您有兴趣，很高兴与kafka进行深入交流，我现在也在深入研究kafka connect的源代码。我的EMAIL：vico.wu1989@gmail.com .