2.2. 在 Apache Kafka 的 Streams 上部署 Debezium

流程

1. 登录 OpenShift 集群。

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

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

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

   • Debezium 连接器存档。
   • 红帽构建的 Apicurio Registry 归档。Apicurio Registry 是一个可选组件。只有在打算将 Avro serialization 与连接器一起使用时，才添加 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.6.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.7.3.Final-redhat-00001/debezium-connector-mysql-2.7.3.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.7.3.Final-redhat-00001/debezium-scripting-2.7.3.Final-redhat-00001.zip 9
             - type: jar
               url: https://repo1.maven.org/maven2/org/apache/groovy/groovy/3.0.11/groovy-3.0.11.jar  10
             - type: jar
               url: https://repo1.maven.org/maven2/org/apache/groovy/groovy-jsr223/3.0.11/groovy-jsr223-3.0.11.jar
             - type: jar
               url: https://repo1.maven.org/maven2/org/apache/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 的更多信息，请参阅 {Name configuringStreamsOpenShift} 中的 Apache Kafka 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 存储库，用于存储连接器工件的文件。Red Hat Maven 存储库中提供了 Debezium 连接器工件。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 所需的。 重要 如果您使用 Streams for Apache Kafka 将连接器插件合并到 Kafka Connect 镜像中，对于每个所需的脚本语言组件 artifacts.url 必须指定 JAR 文件的位置，并且 artifacts.type 的值也必须设置为 jar。无效的值会导致连接器在运行时失败。 要启用将 Apache Groovy 语言与脚本 SMT 搭配使用，示例中的自定义资源会检索以下库的 JAR 文件： groovy groovy-jsr223 （脚本代理） groovy-json （用于解析 JSON 字符串的模块） 作为替代方案，Debezium 脚本 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. mysql-inventory-connector.yaml 文件，为 Debezium 连接器定义 KafkaConnector 自定义资源

       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 主题的前缀，因此该名称在集群中的连接器中必须是唯一的。 如果您将连接器与 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 就绪后，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

   指定主题名称的格式与步骤 1 中返回的 oc describe 命令相同，例如 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":"int64","optional":false,"field":"ts_us"},{"type":"int64","optional":false,"field":"ts_ns"},{"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":"int64","optional":true,"field":"ts_us"},{"type":"int64","optional":true,"field":"ts_ns"},{"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.7.3.Final-redhat-00001","connector":"mysql","name":"inventory-connector-mysql","ts_ms":1638985247805,"ts_us":1638985247805000000,"ts_ns":1638985247805000000,"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,"ts_us":1638985247805102,"ts_ns":1638985247805102588,"transaction":null}}

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