[TOC]
<br >
# **JDBC Connector**
*****
<br >
## **安装 JDBC Connector**
1. 在 [Confluent Hub](https://www.confluent.io/hub/confluentinc/kafka-connect-jdbc) 上下载 ZIP 文件。
2. 解压缩 ZIP 文件并且将内容复制到你想要的目录。例如，你可以创建一个名为`share/kafka/plugins/`目录，并将 Connector 插件内容复制到该目录下。
```shell
# mkdir -p /usr/local/share/kafka/plugins/
# unzip confluentinc-kafka-connect-jdbc-5.4.0.zip
# chown -R kafka:oper /usr/local/share/kafka/
```
3. 将这个路径添加到你的 Kafka Connect 属性文件中,Kafka Connect 使用插件路径查找插件，一个插件路径是一个以逗号分隔的目录列表。
```
$ vim config/connect-distributed.properties
plugin.path=/usr/local/share/kafka/plugins/confluentinc-kafka-connect-jdbc-5.4.0/lib
```
4. 使用配置文件启动 Connet Worker，Connect 将通过这些插件发现所有的连接器。
```shell
$ nohup bin/connect-distributed.sh config/connect-distributed.properties &
... ...
[2020-01-19 15:09:33,463] INFO [Worker clientId=connect-1, groupId=connect-cluster] Starting connectors and tasks using config offset 35 (org.apache.kafka.connect.runtime.distributed.DistributedHerder:1000)
[2020-01-19 15:09:33,463] INFO [Worker clientId=connect-1, groupId=connect-cluster] Finished starting connectors and tasks (org.apache.kafka.connect.runtime.distributed.DistributedHerder:1021)
```
5. 在运行 Connect Worker 的每台机器上重复这些步骤，以保证每个 Worker 上的连接器都已经被激活。
6. 验证插件是否被成功加载。
```shell
$ curl http://localhost:8083/connector-plugins | jq
[
{
"class": "io.confluent.connect.jdbc.JdbcSinkConnector",
"type": "sink",
"version": "5.4.0"
},
{
"class": "io.confluent.connect.jdbc.JdbcSourceConnector",
"type": "source",
"version": "5.4.0"
},
{
"class": "org.apache.kafka.connect.file.FileStreamSinkConnector",
"type": "sink",
"version": "2.3.1"
},
{
"class": "org.apache.kafka.connect.file.FileStreamSourceConnector",
"type": "source",
"version": "2.3.1"
}
]
```
> 你应该可以看到`JdbcSourceConnector`、`JdbcSinkConnector`两个 Connector 插件
7. 列出所有活动的 Connector：
```
$ curl localhost:8083/connectors
[]
```
<br >
## **安装 JDBC 驱动程序**
1. 查找适合的 JDBC 4.0 的驱动 JAR 包。
* [JDBC drivers for Oracle](https://www.oracle.com/technetwork/database/application-development/jdbc/downloads/index.html)
2. 将这些 JAR 包移动到`share/java/kafka-connect-jdbc`目录。
```shell
# mkdir -p /usr/local/share/java/kafka-connect-jdbc/
# chown -R lemon:oper /usr/local/share/java/kafka-connect-jdbc/
```
3. 设置环境变量：
```
$ export CLASSPATH=.:/usr/local/share/java/kafka-connect-jdbc/ojdbc8.jar
```
4. 重启 Connect Worker 节点。例如：
```shell
$ jps -l
87537 org.apache.kafka.connect.cli.ConnectDistributed
$ kill -9 87537
$ nohup bin/connect-distributed.sh config/connect-distributed.properties &
```
5. 验证插件是否被成功加载。
```shell
$ curl http://localhost:8083/connector-plugins | jq
```
