2.3. 在 Red Hat Enterprise Linux 中使用 AMQ Streams 部署 Debezium

流程

1. 从 Red Hat Integration 下载站点下载 Debezium 连接器或连接器。例如，要将 Debezium 与 MySQL 数据库搭配使用，请下载 Debezium 2.3.4 MySQL Connector。

2. 在部署 AMQ Streams 的 Red Hat Enterprise Linux 主机上，打开终端窗口并在 /opt/kafka 中创建 connector-plugins 目录（如果它尚不存在）：

   $ sudo mkdir /opt/kafka/connector-plugins

3. 输入以下命令将您下载到 /opt/kafka/connector-plugins 目录的 Debezium 连接器存档的内容。

   $ sudo unzip debezium-connector-mysql-2.3.4.Final.zip -d /opt/kafka/connector-plugins

4. 对您要安装的每个连接器重复步骤 1 -3。

5. 在一个终端窗口中，以 kafka 用户身份进行登录：

   $ su - kafka
   $ Password:

6. 如果正在运行，停止 Kafka Connect 进程。

   1. 输入以下命令检查 Kafka Connect 是否在分布式模式下运行：

      $ jcmd | grep ConnectDistributed

      如果进程正在运行，命令会返回进程 ID，例如：

      18514 org.apache.kafka.connect.cli.ConnectDistributed /opt/kafka/config/connect-distributed.properties

   2. 使用进程 ID 输入 kill 命令来停止该进程，例如：

      $ kill 18514

7. 编辑 /opt/kafka/config/ 中的 connect-distributed.properties 文件，并将 plugin.path 的值设置为 Debezium 连接器插件的父目录的位置：

   plugin.path=/opt/kafka/connector-plugins

8. 以分布式模式启动 Kafka 连接。

   $ /opt/kafka/bin/connect-distributed.sh /opt/kafka/config/connect-distributed.properties

9. 在 Kafka Connect 运行后，使用 Kafka Connect API 来注册连接器。
   输入 curl 命令，以提交 POST 请求，将您在 第 2.2 节 “规划 Debezium 连接器配置” 中指定的连接器配置 JSON 发送到 localhost:8083/connectors 的 Kafka Connect REST API 端点。
   例如：

   curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" localhost:8083/connectors/ \
   -d '{"name": "inventory-connector", "config": \
   { "connector.class": "io.debezium.connector.mysql.MySqlConnector", \
   "tasks.max": "1", \
   "database.hostname": "mysql", \
   "database.port": "3306", \
   "database.user": "debezium", \
   "database.password": "dbz", \
   "database.server.id": "184054", \
   "topic.prefix": "dbserver1", \
   "table.include.list": "public.inventory", \
   "schema.history.internal.kafka.bootstrap.servers": "kafka:9092", \
   "schema.history.internal.kafka.topic": "dbhistory.inventory" } }'

   要注册多个连接器，请为每个连接器提交一个单独的请求。

10. 重启 Kafka Connect 来实现您的更改。

    当 Kafka Connect 启动时，它会从 connector-plugins 目录中加载配置的 Debezium 连接器。

    完成配置后，部署的连接器连接到源数据库，并为每个插入、更新或删除的行或文档生成事件。

11. 为每个 Kafka Connect worker 节点重复步骤 5-10。