3.2. 部署 Kafka Connect

流程

1. 登录 OpenShift 集群，再创建或打开一个项目，如 debezium。

2. 为连接器创建 Debezium KafkaConnect 自定义资源(CR)，或修改现有的资源。
   以下示例显示了一个 dbz-connect.yaml 文件的摘录，该文件描述了 KafkaConnect 自定义资源。
   metadata.annotations 和 spec.build 属性是必需的。

   例 3.1. 定义包含 Debezium 连接器的 KafkaConnect 自定义资源的 dbz-connect.yaml 文件

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaConnect
   metadata:
     name: my-connect-cluster
     annotations:
       strimzi.io/use-connector-resources: "true" 1
   spec:
     replicas: 1
     version: 3.5.0
     build: 2
       output: 3
         type: imagestream  4
         image: debezium-streams-connect:latest
       plugins: 5
         - name: debezium-connector-mysql
           artifacts:
             - type: zip 6
               url: https://maven.repository.redhat.com/ga/io/debezium/debezium-connector-mysql/2.3.4.Final-redhat-00001/debezium-connector-mysql-2.3.4.Final-redhat-00001-plugin.zip 7
     bootstrapServers: my-cluster-kafka-bootstrap:9093
   
   ...

   表 3.1. Kafka Connect 配置设置的描述

   项	描述
   1	将 strimzi.io/use-connector-resources 注解设置为 "true"，使 Cluster Operator 使用 KafkaConnector 资源在此 Kafka Connect 集群中配置连接器。
   2	spec.build 配置指定在镜像中存储构建镜像的位置，并列出要在镜像中包含的插件，以及插件工件的位置。
   3	build.output 指定存储新构建镜像的 registry。
   4	指定镜像输出的名称和镜像名称。output.type 的有效值是要推送到 Docker Hub 或 Quay 等容器 registry 的有效值，或 镜像流 以将镜像推送到内部 OpenShift ImageStream。要使用 ImageStream，必须将 ImageStream 资源部署到集群中。有关在 KafkaConnect 配置中指定 build.output 的更多信息，请参阅 AMQ Streams Build schema 文档。
   5	plugins 配置列出了您要包含在 Kafka Connect 镜像中的所有连接器。对于列表中的每个条目，指定一个插件名称，以及有关构建连接器所需的工件的信息。另外，对于每个连接器插件，您还可以包含可用于连接器的其他组件。例如，您可以添加 Service Registry 工件或 Debezium 脚本组件。
   6	artifacts.type 的值指定在 artifacts.url 中指定的工件类型。有效类型为 zip、tgz 或 jar。Debezium 连接器存档以 .zip 文件格式提供。JDBC 驱动程序文件采用 .jar 格式。类型 值必须与 url 字段中引用的文件类型匹配。
   7	artifacts.url 的值指定 HTTP 服务器的地址，如 Maven 存储库，用于存储连接器工件的文件。OpenShift 集群必须有权访问指定的服务器。

3. 输入以下命令将 KafkaConnect 构建规格应用到 OpenShift 集群：

   oc create -f dbz-connect.yaml

   根据自定义资源中指定的配置，AMQ Streams Operator 准备要部署的 Kafka Connect 镜像。
   构建完成后，Operator 将镜像推送到指定的 registry 或 ImageStream，并启动 Kafka Connect 集群。集群中提供了您在配置中列出的连接器工件。

4. 创建一个 KafkaConnector 资源来定义 MySQL 连接器的实例。
   例如，创建以下 KafkaConnector CR，并将它保存为 debezium-inventory-connector.yaml

   例 3.2. 为 Debezium 连接器定义 KafkaConnector 自定义资源的 mysql-inventory-connector.yaml 文件

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaConnector
   metadata:
     labels:
       strimzi.io/cluster: my-connect-cluster
     name: inventory-connector 1
   spec:
     class: io.debezium.connector.mysql.MySqlConnector 2
     tasksMax: 1  3
     config:  4
       database.hostname: mysql 5
       database.port: 3306   6
       database.user: debezium  7
       database.password: dbz  8
       database.server.id: 184054
       topic.prefix: dbserver1  9
       table.include.list: inventory.*  10
       schema.history.internal.kafka.bootstrap.servers: 'my-cluster-kafka-bootstrap:9092' 11
       schema.history.internal.kafka.topic: schema-changes.inventory 12

   表 3.2. 连接器配置设置的描述

   项	描述
   1	使用 Kafka Connect 集群注册的连接器名称。
   2	连接器类的名称。
   3	任何时候都只能运行一个任务。使用单一连接器任务来确保正确的顺序和事件处理，因为 MySQL 连接器读取 MySQL 服务器的 binlog。Kafka Connect 服务使用连接器来启动一个或多个任务来完成工作。它会在 Kafka Connect 服务集群中自动分发正在运行的任务。如果服务停止或崩溃，任务将重新分发到运行的服务。
   4	连接器的配置。
   5	MySQL 数据库实例的主机名或地址。
   6	数据库实例的端口号。
   7	Debezium 通过它连接到数据库的用户帐户名称。
   8	Debezium 用于连接到数据库用户帐户的密码。
   9	MySQL 服务器或集群的主题前缀。此字符串前缀连接器将事件记录发送到的每个 Kafka 主题的名称。
   10	连接器捕获更改事件的表列表。连接器仅在 清单 表中发生时才会检测更改。
   11	连接器用来写入和恢复 DDL 语句到数据库 schema 历史记录主题的 Kafka 代理列表。这与连接器发送更改事件记录的代理相同。重启后，连接器会在连接器恢复读取时恢复 binlog 中存在的数据库模式。
   12	数据库架构历史记录主题的名称。本主题仅用于内部使用，不应供消费者使用。

5. 运行以下命令来创建连接器资源：

   oc create -n <namespace> -f <kafkaConnector>.yaml

   例如，

   oc create -n debezium -f mysql-inventory-connector.yaml

   连接器注册到 Kafka Connect 集群，并开始针对 KafkaConnector CR 中的 spec.config.database.dbname 指定的数据库运行。连接器 pod 就绪后，Debebe 正在运行。