2.2. AMQ Streams 上的 Debezium 部署

流程

1. 登录 OpenShift 集群。

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

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

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

   • Debezium 连接器存档。
   • 红帽构建的 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.3.1
     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.1.4.Final-redhat-00001/debezium-connector-mysql-2.1.4.Final-redhat-00001-plugin.zip  7
             - type: zip
               url: https://maven.repository.redhat.com/ga/io/apicurio/apicurio-registry-distro-connect-converter/2.3.0.Final-redhat-<build-number>/apicurio-registry-distro-connect-converter-2.3.0.Final-redhat-<build-number>.zip  8
             - type: zip
               url: https://maven.repository.redhat.com/ga/io/debezium/debezium-scripting/2.1.4.Final-redhat-00001/debezium-scripting-2.1.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
   
     ...

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

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

   例 2.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
   
           ...

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

   项	描述
   1	使用 Kafka Connect 集群注册的连接器名称。
   2	连接器类的名称。
   3	可同时运行的任务数量。
   4	连接器的配置。
   5	主机数据库实例的地址。
   6	数据库实例的端口号。
   7	Debezium 用于连接到数据库的帐户名称。
   8	Debezium 用于连接到数据库用户帐户的密码。
   9	连接器的唯一数字 ID。
   10	数据库实例或集群的主题前缀。 指定的名称只能从字母数字字符或下划线括起。 由于主题前缀用作从这个连接器接收更改事件的任何 Kafka 主题的前缀，所以名称必须在集群中的连接器之间唯一。 此命名空间也用于相关的 Kafka Connect 模式的名称，如果您将连接器与 Avro 连接器集成，则对应的 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 就绪后，Debezium 正在运行。

流程

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

        该命令返回类似以下输出的状态信息：

        例 2.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

        该命令返回类似以下输出的状态信息：

        例 2.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。

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

   例 2.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.1.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。