5.5. 部署 Debezium MySQL 连接器

流程

1. 登录 OpenShift 集群。

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

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

   在以下示例中，自定义资源被配置为下载以下工件：

   • Debezium MySQL 连接器存档。
   • 红帽构建的 Apicurio Registry 存档。Apicurio Registry 是一个可选组件。只有在打算将 Avro 序列化与连接器搭配使用时，才添加 Apicurio Registry 组件。
   • Debezium 脚本 SMT 归档以及您要与 Debezium 连接器一起使用的关联脚本引擎。SMT 归档和脚本语言依赖项是可选组件。只有在打算使用 Debezium 的基于内容的路由 SMT 或 过滤 SMT 时，才添加这些组件。

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaConnect
   metadata:
     name: debezium-kafka-connect-cluster
     annotations:
       strimzi.io/use-connector-resources: "true" 1
   spec:
     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
             - type: zip
               url: https://maven.repository.redhat.com/ga/io/apicurio/apicurio-registry-distro-connect-converter/2.4.4.Final-redhat-<build-number>/apicurio-registry-distro-connect-converter-2.4.4.Final-redhat-<build-number>.zip  8
             - type: zip
               url: https://maven.repository.redhat.com/ga/io/debezium/debezium-scripting/2.3.4.Final-redhat-00001/debezium-scripting-2.3.4.Final-redhat-00001.zip 9
             - type: jar
               url: https://repo1.maven.org/maven2/org/codehaus/groovy/groovy/3.0.11/groovy-3.0.11.jar  10
             - type: jar
               url: https://repo1.maven.org/maven2/org/codehaus/groovy/groovy-jsr223/3.0.11/groovy-jsr223-3.0.11.jar
             - type: jar
               url: https://repo1.maven.org/maven2/org/codehaus/groovy/groovy-json3.0.11/groovy-json-3.0.11.jar
   
     bootstrapServers: debezium-kafka-cluster-kafka-bootstrap:9093
   
     ...

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

   项	描述
   1	将 strimzi.io/use-connector-resources 注解设置为 "true"，使 Cluster Operator 使用 KafkaConnector 资源在此 Kafka Connect 集群中配置连接器。
   2	spec.build 配置指定在镜像中存储构建镜像的位置，并列出要在镜像中包含的插件，以及插件工件的位置。
   3	build.output 指定存储新构建镜像的 registry。
   4	指定镜像输出的名称和镜像名称。output.type 的有效值是 要推送到 容器 registry （如 Docker Hub 或 Quay）或 镜像流 的有效值，以将镜像推送到内部 OpenShift ImageStream。要使用 ImageStream，必须将 ImageStream 资源部署到集群中。有关在 KafkaConnect 配置中指定 build.output 的更多信息，请参阅在 OpenShift 中配置 AMQ Streams 中的 AMQ Streams Build schema 参考。
   5	plugins 配置列出了您要包含在 Kafka Connect 镜像中的所有连接器。对于列表中的每个条目，指定一个插件名称，以及有关构建连接器所需的工件的信息。另外，对于每个连接器插件，您还可以包含可用于连接器的其他组件。例如，您可以添加 Service Registry 工件或 Debezium 脚本组件。
   6	artifacts.type 的值指定在 artifacts.url 中指定的工件类型。有效类型为 zip、tgz 或 jar。Debezium 连接器存档以 .zip 文件格式提供。类型 值必须与 url 字段中引用的文件类型匹配。
   7	artifacts.url 的值指定 HTTP 服务器的地址，如 Maven 存储库，用于存储连接器工件的文件。Debezium 连接器工件在 Red Hat Maven 存储库中提供。OpenShift 集群必须有权访问指定的服务器。
   8	（可选）指定用于下载 Apicurio Registry 组件的工件 类型和 url。包含 Apicurio Registry 工件，只有在您希望连接器使用 Apache Avro 来序列化带有红帽构建的 Apicurio Registry 的值，而不是使用默认的 JSON 转换程序时。
   9	（可选）指定 Debezium 脚本 SMT 归档的工件 类型和 url，以用于 Debezium 连接器。只有在打算使用 Debezium 的基于内容的路由 SMT 或 过滤 SMT 时才包括脚本 SMT。要使用脚本 SMT，您必须部署 JSR 223 兼容脚本实现，如 groovy。
   10	（可选）指定 JSR 223 兼容脚本实施的 JAR 文件的工件 类型和 url，这是 Debezium 脚本 SMT 所需的。 重要 如果使用 AMQ Streams 将连接器插件合并到 Kafka Connect 镜像中，每个所需的脚本语言 工件。url 必须指定 JAR 文件的位置，并且 artifacts.type 的值也必须设置为 jar。无效的值会导致连接器在运行时失败。 要启用带有脚本 SMT 的 Apache Groovy 语言，示例中的自定义资源会为以下库检索 JAR 文件： groovy Groovy-jsr223 （指定代理） groovy-json （解析 JSON 字符串的模块） 作为替代方案，Debebe Debezium 脚本 SMT 也支持使用 JSR 223 实现 GraalVM JavaScript。

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

   oc create -f dbz-connect.yaml

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

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

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

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

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

   项	描述
   1	使用 Kafka Connect 集群注册的连接器名称。
   2	连接器类的名称。
   3	可以同时操作的任务数量。
   4	连接器的配置。
   5	主机数据库实例的地址。
   6	数据库实例的端口号。
   7	Debezium 用于连接到数据库的帐户名称。
   8	Debezium 用于连接到数据库用户帐户的密码。
   9	连接器的唯一数字 ID。
   10	数据库实例或集群的主题前缀。 指定的名称只能由字母数字字符或下划线组成。 因为主题前缀被用作从这个连接器接收更改事件的任何 Kafka 主题的前缀，所以该名称在集群中的连接器之间必须是唯一的。 如果连接器与 Avro 连接器集成，则此命名空间也用于相关 Kafka Connect 模式的名称，以及相应 Avro 模式的命名空间。
   11	连接器捕获更改事件的表列表。

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 正在运行。

流程

1. 为 Kafka Connect 创建 Debezium MySQL 容器：

   1. 创建一个使用 registry.redhat.io/amq-streams-kafka-35-rhel8:2.5.0 的 Dockerfile 作为基础镜像。例如，在终端窗口中输入以下命令：

      cat <<EOF >debezium-container-for-mysql.yaml 1
      FROM registry.redhat.io/amq-streams-kafka-35-rhel8:2.5.0
      USER root:root
      RUN mkdir -p /opt/kafka/plugins/debezium 2
      RUN cd /opt/kafka/plugins/debezium/ \
      && curl -O 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 \
      && unzip debezium-connector-mysql-2.3.4.Final-redhat-00001-plugin.zip \
      && rm debezium-connector-mysql-2.3.4.Final-redhat-00001-plugin.zip
      RUN cd /opt/kafka/plugins/debezium/
      USER 1001
      EOF

      项	描述
      1	您可以指定您想要的任何文件名。
      2	指定 Kafka Connect 插件目录的路径。如果您的 Kafka Connect 插件目录位于不同的位置，请将此路径替换为目录的实际路径。

      该命令在当前目录中创建一个名为 debezium-container-for-mysql.yaml 的 Dockerfile。

   2. 从您在上一步中创建的 debezium-container-for-mysql.yaml Docker 文件中构建容器镜像。在包含文件的目录中，打开终端窗口并输入以下命令之一：

      podman build -t debezium-container-for-mysql:latest .

      docker build -t debezium-container-for-mysql:latest .

      前面的命令使用名称 debezium-container-for-mysql 构建容器镜像。

   3. 将自定义镜像推送到容器 registry，如 quay.io 或内部容器 registry。容器 registry 必须可供您要部署镜像的 OpenShift 实例使用。输入以下命令之一：

      podman push <myregistry.io>/debezium-container-for-mysql:latest

      docker push <myregistry.io>/debezium-container-for-mysql:latest

   4. 创建新的 Debezium MySQL KafkaConnect 自定义资源(CR)。例如，创建一个名为 dbz-connect.yaml 的 KafkaConnect CR，用于指定 注解和 镜像 属性。以下示例显示了一个 dbz-connect.yaml 文件的摘录，该文件描述了 KafkaConnect 自定义资源。
      

      apiVersion: kafka.strimzi.io/v1beta2
      kind: KafkaConnect
      metadata:
        name: my-connect-cluster
        annotations:
          strimzi.io/use-connector-resources: "true" 1
      spec:
        #...
        image: debezium-container-for-mysql  2
      
        ...

      项	描述
      1	metadata.annotations 表示 KafkaConnector 资源用于配置在这个 Kafka Connect 集群中使用的 Cluster Operator。
      2	spec.image 指定您创建的镜像的名称，以运行 Debezium 连接器。此属性覆盖 Cluster Operator 中的 STRIMZI_DEFAULT_KAFKA_CONNECT_IMAGE 变量。

   5. 输入以下命令将 KafkaConnect CR 应用到 OpenShift Kafka Connect 环境：

      oc create -f dbz-connect.yaml

      该命令添加了一个 Kafka Connect 实例，用于指定您为运行 Debezium 连接器而创建的镜像的名称。

2. 创建一个 KafkaConnector 自定义资源来配置 Debezium MySQL 连接器实例。

   您可以在 .yaml 文件中配置 Debezium MySQL 连接器，该文件指定连接器的配置属性。连接器配置可能指示 Debezium 为 schema 和表的子集生成事件，或者可能会设置属性，以便 Debezium 忽略、掩码或截断敏感、太大或不需要的指定列中的值。

   以下示例配置了一个 Debezium 连接器，它连接到 MySQL 主机 192.168.99.100，使用端口 3306，它会捕获 inventory 数据库的更改。dbserver1 是服务器的逻辑名称。

   MySQL inventory-connector.yaml

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

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

   项	描述
   1	连接器的名称。
   2	任何时候都只能运行一个任务。因为 MySQL 连接器使用单一连接器任务读取 MySQL 服务器的 binlog 可确保正确的顺序和事件处理。Kafka Connect 服务使用连接器来启动一个或多个完成工作的任务，并在 Kafka Connect 服务集群中自动分发正在运行的任务。如果有任何服务停止或崩溃，则这些任务将重新分发到运行的服务。
   3	连接器的配置。
   4	数据库主机，这是运行 MySQL 服务器的容器的名称(mysql)。
   5	连接器的唯一 ID。
   6	MySQL 服务器或集群的主题前缀。此名称用作接收更改事件记录的所有 Kafka 主题的前缀。
   7	连接器只捕获 inventory 表中的更改。
   8	此连接器将用来写入和恢复 DDL 语句到数据库 schema 历史记录主题的 Kafka 代理列表。重启时，连接器会在连接器开始读取时恢复 binlog 中存在的数据库的模式。
   9	数据库架构历史记录主题的名称。本主题仅用于内部使用，不应供消费者使用。

3. 使用 Kafka Connect 创建连接器实例。例如，如果您将 KafkaConnector 资源保存在 inventory-connector.yaml 文件中，您将运行以下命令：

   oc apply -f inventory-connector.yaml

   前面的命令注册 inventory-connector，连接器开始针对 KafkaConnector CR 中定义的 inventory 数据库运行。

流程

1. 使用以下方法之一检查 KafkaConnector 资源的状态：

   • 在 OpenShift Container Platform Web 控制台中：

     1. 导航到 Home Search。
     2. 在 Search 页面中，点 Resources 打开 Select Resource 框，然后键入 KafkaConnector。
     3. 在 KafkaConnectors 列表中，点您要检查的连接器的名称，如 inventory-connector-mysql。
     4. 在 Conditions 部分，验证 Type 和 Status 列中的值是否已设置为 Ready 和 True。

   • 在终端窗口中：

     1. 使用以下命令：

        oc describe KafkaConnector <connector-name> -n <project>

        例如，

        oc describe KafkaConnector inventory-connector-mysql -n debezium

        该命令返回类似以下示例的状态信息：

        例 5.3. KafkaConnector 资源状态

        Name:         inventory-connector-mysql
        Namespace:    debezium
        Labels:       strimzi.io/cluster=debezium-kafka-connect-cluster
        Annotations:  <none>
        API Version:  kafka.strimzi.io/v1beta2
        Kind:         KafkaConnector
        
        ...
        
        Status:
          Conditions:
            Last Transition Time:  2021-12-08T17:41:34.897153Z
            Status:                True
            Type:                  Ready
          Connector Status:
            Connector:
              State:      RUNNING
              worker_id:  10.131.1.124:8083
            Name:         inventory-connector-mysql
            Tasks:
              Id:               0
              State:            RUNNING
              worker_id:        10.131.1.124:8083
            Type:               source
          Observed Generation:  1
          Tasks Max:            1
          Topics:
            inventory-connector-mysql.inventory
            inventory-connector-mysql.inventory.addresses
            inventory-connector-mysql.inventory.customers
            inventory-connector-mysql.inventory.geom
            inventory-connector-mysql.inventory.orders
            inventory-connector-mysql.inventory.products
            inventory-connector-mysql.inventory.products_on_hand
        Events:  <none>

2. 验证连接器是否创建了 Kafka 主题：

   • 通过 OpenShift Container Platform Web 控制台。

     1. 导航到 Home Search。
     2. 在 Search 页面中，点 Resources 打开 Select Resource 框，然后键入 KafkaTopic。
     3. 在 KafkaTopics 列表中，点您要检查的主题名称，例如 inventory-connector-mysql.inventory.orders---ac5e98ac6a5d91e04d8ec0dc9078a1ece439081d。
     4. 在 Conditions 部分，验证 Type 和 Status 列中的值是否已设置为 Ready 和 True。

   • 在终端窗口中：

     1. 使用以下命令：

        oc get kafkatopics

        该命令返回类似以下示例的状态信息：

        例 5.4. KafkaTopic 资源状态

        NAME                                                                    CLUSTER               PARTITIONS   REPLICATION FACTOR   READY
        connect-cluster-configs                                                 debezium-kafka-cluster   1            1                    True
        connect-cluster-offsets                                                 debezium-kafka-cluster   25           1                    True
        connect-cluster-status                                                  debezium-kafka-cluster   5            1                    True
        consumer-offsets---84e7a678d08f4bd226872e5cdd4eb527fadc1c6a             debezium-kafka-cluster   50           1                    True
        inventory-connector-mysql--a96f69b23d6118ff415f772679da623fbbb99421                               debezium-kafka-cluster   1            1                    True
        inventory-connector-mysql.inventory.addresses---1b6beaf7b2eb57d177d92be90ca2b210c9a56480          debezium-kafka-cluster   1            1                    True
        inventory-connector-mysql.inventory.customers---9931e04ec92ecc0924f4406af3fdace7545c483b          debezium-kafka-cluster   1            1                    True
        inventory-connector-mysql.inventory.geom---9f7e136091f071bf49ca59bf99e86c713ee58dd5               debezium-kafka-cluster   1            1                    True
        inventory-connector-mysql.inventory.orders---ac5e98ac6a5d91e04d8ec0dc9078a1ece439081d             debezium-kafka-cluster   1            1                    True
        inventory-connector-mysql.inventory.products---df0746db116844cee2297fab611c21b56f82dcef           debezium-kafka-cluster   1            1                    True
        inventory-connector-mysql.inventory.products_on_hand---8649e0f17ffcc9212e266e31a7aeea4585e5c6b5   debezium-kafka-cluster   1            1                    True
        schema-changes.inventory                                                debezium-kafka-cluster   1            1                    True
        strimzi-store-topic---effb8e3e057afce1ecf67c3f5d8e4e3ff177fc55          debezium-kafka-cluster   1            1                    True
        strimzi-topic-operator-kstreams-topic-store-changelog---b75e702040b99be8a9263134de3507fc0cc4017b  debezium-kafka-cluster  1   1    True

3. 检查主题内容。

   • 在终端窗口中输入以下命令：

   oc exec -n <project>  -it <kafka-cluster> -- /opt/kafka/bin/kafka-console-consumer.sh \
   >     --bootstrap-server localhost:9092 \
   >     --from-beginning \
   >     --property print.key=true \
   >     --topic=<topic-name>

   例如，

   oc exec -n debezium  -it debezium-kafka-cluster-kafka-0 -- /opt/kafka/bin/kafka-console-consumer.sh \
   >     --bootstrap-server localhost:9092 \
   >     --from-beginning \
   >     --property print.key=true \
   >     --topic=inventory-connector-mysql.inventory.products_on_hand

   指定主题名称的格式与 oc describe 命令返回的格式与第 1 步中返回，例如 inventory-connector-mysql.inventory.addresses。

   对于主题中的每个事件，命令会返回类似以下示例的信息：

   例 5.5. Debezium 更改事件的内容

   {"schema":{"type":"struct","fields":[{"type":"int32","optional":false,"field":"product_id"}],"optional":false,"name":"inventory-connector-mysql.inventory.products_on_hand.Key"},"payload":{"product_id":101}} {"schema":{"type":"struct","fields":[{"type":"struct","fields":[{"type":"int32","optional":false,"field":"product_id"},{"type":"int32","optional":false,"field":"quantity"}],"optional":true,"name":"inventory-connector-mysql.inventory.products_on_hand.Value","field":"before"},{"type":"struct","fields":[{"type":"int32","optional":false,"field":"product_id"},{"type":"int32","optional":false,"field":"quantity"}],"optional":true,"name":"inventory-connector-mysql.inventory.products_on_hand.Value","field":"after"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"version"},{"type":"string","optional":false,"field":"connector"},{"type":"string","optional":false,"field":"name"},{"type":"int64","optional":false,"field":"ts_ms"},{"type":"string","optional":true,"name":"io.debezium.data.Enum","version":1,"parameters":{"allowed":"true,last,false"},"default":"false","field":"snapshot"},{"type":"string","optional":false,"field":"db"},{"type":"string","optional":true,"field":"sequence"},{"type":"string","optional":true,"field":"table"},{"type":"int64","optional":false,"field":"server_id"},{"type":"string","optional":true,"field":"gtid"},{"type":"string","optional":false,"field":"file"},{"type":"int64","optional":false,"field":"pos"},{"type":"int32","optional":false,"field":"row"},{"type":"int64","optional":true,"field":"thread"},{"type":"string","optional":true,"field":"query"}],"optional":false,"name":"io.debezium.connector.mysql.Source","field":"source"},{"type":"string","optional":false,"field":"op"},{"type":"int64","optional":true,"field":"ts_ms"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"id"},{"type":"int64","optional":false,"field":"total_order"},{"type":"int64","optional":false,"field":"data_collection_order"}],"optional":true,"field":"transaction"}],"optional":false,"name":"inventory-connector-mysql.inventory.products_on_hand.Envelope"},"payload":{"before":null,"after":{"product_id":101,"quantity":3},"source":{"version":"2.3.4.Final-redhat-00001","connector":"mysql","name":"inventory-connector-mysql","ts_ms":1638985247805,"snapshot":"true","db":"inventory","sequence":null,"table":"products_on_hand","server_id":0,"gtid":null,"file":"mysql-bin.000003","pos":156,"row":0,"thread":null,"query":null},"op":"r","ts_ms":1638985247805,"transaction":null}}

   在前面的示例中，有效负载 值显示连接器快照从表 inventory.products_on_hand 生成读取(op" ="r")事件。product_id 记录的 "before" 状态为 null，表示该记录不存在之前的值。"after" 状态对于 product_id 为 101 的项目的 quantity 显示为 3。