Kafka Connect 作为service提供了 REST API 服务，用来：

• 获取 Kafka Connect 状态

• 管理 Kafka Connect 配置

• Kafka Connect 集群内部通信

常用命令如下：

• curl -s <Kafka Connect Worker URL>:8083/ | jq
  获取 Connect Worker 信息

• curl -s <Kafka Connect Worker URL>:8083/connector-plugins | jq
  列出 Connect Worker 上所有 Connector

• curl -s <Kafka Connect Worker URL>:8083/connectors/<Connector名字>/tasks | jq
  获取 Connector 上 Task 以及相关配置的信息

• curl -s <Kafka Connect Worker URL>:8083/connectors/<Connector名字>/status | jq
  获取 Connector 状态信息

• curl -s <Kafka Connect Worker URL>:8083/connectors/<Connector名字>/config | jq
  获取 Connector 配置信息

• curl -s -X PUT <Kafka Connect Worker URL>:8083/connectors/<Connector名字>/pause
  暂停 Connector

• curl -s -X PUT <Kafka Connect Worker URL>:8083/connectors/<Connector名字>/resume
  重启 Connector

• curl -s -X DELETE <Kafka Connect Worker URL>:8083/connectors/<Connector名字>
  删除 Connector

curl -s -X POST -H "Content-Type: application/json" --data 
 '{"name": "<Connector名字>", 
   "config":
    {"connector.class":"org.apache.kafka.connect.file.FileStreamSourceConnector",
    "key.converter.schemas.enable":"true",
    "file":"demo-file.txt",
    "tasks.max":"1",
    "value.converter.schemas.enable":"true",
    "name":"file-stream-demo-distributed",
    "topic":"demo-distributed",
    "value.converter":"org.apache.kafka.connect.json.JsonConverter",
    "key.converter":"org.apache.kafka.connect.json.JsonConverter"}
 }' 
http://<Kafka Connect Worker URL>:8083/connectors | jq

创建新 Connector （以FileStreamSourceConnector举例）

curl -s -X PUT -H "Content-Type: application/json" --data 
'{"connector.class":"org.apache.kafka.connect.file.FileStreamSourceConnector",
"key.converter.schemas.enable":"true",
"file":"demo-file.txt",
"tasks.max":"2",
"value.converter.schemas.enable":"true",
"name":"file-stream-demo-distributed",
"topic":"demo-2-distributed",
"value.converter":"org.apache.kafka.connect.json.JsonConverter",
"key.converter":"org.apache.kafka.connect.json.JsonConverter"}' 
<Kafka Connect Worker URL>:8083/connectors/file-stream-demo-distributed/config | jq

更新 Connector配置 （以FileStreamSourceConnector举例）