4.5. 部署 Debezium MongoDB 连接器

流程

1. 登录 OpenShift 集群。

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

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

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

   • Debezium MongoDB 连接器存档。
   • Service Registry 归档。Service Registry 是一个可选组件。只有在打算在连接器中使用 Avro 序列化时，才添加 Service 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-mongodb
           artifacts:
             - type: zip 6
               url: https://maven.repository.redhat.com/ga/io/debezium/debezium-connector-mongodb/2.1.4.Final-redhat-00001/debezium-connector-mongodb-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
   
     ...

   表 4.10. 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	（可选）指定下载 Service Registry 组件的工件 类型和 url。包括 Service Registry 工件，只有在您希望连接器使用 Apache Avro 与 Service 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，并将它保存为 mongodb-inventory-connector.yaml

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

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaConnector
   metadata:
     labels:
       strimzi.io/cluster: debezium-kafka-connect-cluster
     name: inventory-connector-mongodb 1
   spec:
     class: io.debezium.connector.mongodb.MongoDbConnector 2
     tasksMax: 1  3
     config:  4
       mongodb.hosts: rs0/192.168.99.100:27017 5
       mongodb.user: debezium  6
       mongodb.password: dbz  7
       topic.prefix: inventory-connector-mongodb 8
       collection.include.list: inventory[.]*  9

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

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

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

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

   例如，

   oc create -n debezium -f {context}-inventory-connector.yaml

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

流程

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

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

      cat <<EOF >debezium-container-for-mongodb.yaml 1
      FROM registry.redhat.io/amq7/amq-streams-kafka-32-rhel8:2.2.0-12
      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-mongodb/2.1.4.Final-redhat-00001/debezium-connector-mongodb-2.1.4.Final-redhat-00001-plugin.zip \
      && unzip debezium-connector-mongodb-2.1.4.Final-redhat-00001-plugin.zip \
      && rm debezium-connector-mongodb-2.1.4.Final-redhat-00001-plugin.zip
      RUN cd /opt/kafka/plugins/debezium/
      USER 1001
      EOF

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

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

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

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

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

      上述命令使用名称 debezium-container-for-mongodb 构建容器镜像。

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

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

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

   4. 创建新的 Debezium MongoDB 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-mongodb  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 MongoDB 连接器实例。

   您可以在 .yaml 文件中配置 Debezium MongoDB 连接器，该文件指定连接器的配置属性。连接器配置可能会指示 Debezium 为 MongoDB 副本集或分片集群的子集生成更改事件。另外，您可以设置过滤不需要的集合的属性。

   以下示例配置了 Debezium 连接器，它连接到在 192.168.99.100 上的端口 27017 处连接到 MongoDB 副本集 rs0，并捕获 清单 集合中发生的更改。inventory-connector-mongodb 是副本集的逻辑名称。

   MongoDB inventory-connector.yaml

   apiVersion: kafka.strimzi.io/v1beta2
     kind: KafkaConnector
     metadata:
       name: inventory-connector-mongodb 1
       labels: strimzi.io/cluster: my-connect-cluster
     spec:
       class: io.debezium.connector.mongodb.MongoDbConnector 2
       config:
        mongodb.connection.string: mongodb://192.168.99.100:27017/?replicaSet=rs0 3
        topic.prefix: inventory-connector-mongodb 4
        collection.include.list: inventory[.]* 5

   1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1

   用于在 Kafka Connect 中注册连接器的名称。

   2 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2

   MongoDB 连接器类的名称。

   3 3 3 3 3 3 3 3 3 3 3 3 3 3 3 3 3 3 3 3 3

   用于连接到 MongoDB 副本集的主机地址。

   4 4 4 4 4 4 4 4 4 4 4 4 4 4 4 4 4 4 4 4 4

   MongoDB 副本集的逻辑名称，它组成了生成事件的命名空间，并在使用 Avro converter 时，用来写入的 Kafka 主题、Kafka Connect 模式名称和相应 Avro 模式的命名空间中使用。

   5 5 5 5 5 5 5 5 5 5 5 5 5 5 5 5 5 5

   与要监控的所有集合的集合命名空间（如 <dbName>.<collectionName>）匹配的正则表达式列表。

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

   oc apply -f inventory-connector.yaml

   上述命令注册 inventory-connector，连接器开始针对 KafkaConnector CR 中定义的 清单 集合运行。

流程

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

   • 在 OpenShift Container Platform Web 控制台中：

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

   • 在一个终端窗口中：

     1. 使用以下命令：

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

        例如，

        oc describe KafkaConnector inventory-connector-mongodb -n debezium

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

        例 4.3. KafkaConnector 资源状态

        Name:         inventory-connector-mongodb
        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-mongodb
            Tasks:
              Id:               0
              State:            RUNNING
              worker_id:        10.131.1.124:8083
            Type:               source
          Observed Generation:  1
          Tasks Max:            1
          Topics:
            inventory-connector-mongodb.inventory
            inventory-connector-mongodb.inventory.addresses
            inventory-connector-mongodb.inventory.customers
            inventory-connector-mongodb.inventory.geom
            inventory-connector-mongodb.inventory.orders
            inventory-connector-mongodb.inventory.products
            inventory-connector-mongodb.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-mongodb.inventory.orders--ac5e98ac6a5d91e04d8ec0dc9078a1ece439081d。
     4. 在 Conditions 部分中，验证 Type 和 Status 列中的值是否已设置为 Ready 和 True。

   • 在一个终端窗口中：

     1. 使用以下命令：

        oc get kafkatopics

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

        例 4.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-mongodb--a96f69b23d6118ff415f772679da623fbbb99421                               debezium-kafka-cluster   1            1                    True
        inventory-connector-mongodb.inventory.addresses---1b6beaf7b2eb57d177d92be90ca2b210c9a56480          debezium-kafka-cluster   1            1                    True
        inventory-connector-mongodb.inventory.customers---9931e04ec92ecc0924f4406af3fdace7545c483b          debezium-kafka-cluster   1            1                    True
        inventory-connector-mongodb.inventory.geom---9f7e136091f071bf49ca59bf99e86c713ee58dd5               debezium-kafka-cluster   1            1                    True
        inventory-connector-mongodb.inventory.orders---ac5e98ac6a5d91e04d8ec0dc9078a1ece439081d             debezium-kafka-cluster   1            1                    True
        inventory-connector-mongodb.inventory.products---df0746db116844cee2297fab611c21b56f82dcef           debezium-kafka-cluster   1            1                    True
        inventory-connector-mongodb.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-mongodb.inventory.products_on_hand

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

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

   例 4.5. Debezium 更改事件的内容

   {"schema":{"type":"struct","fields":[{"type":"int32","optional":false,"field":"product_id"}],"optional":false,"name":"inventory-connector-mongodb.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-mongodb.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-mongodb.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.mongodb.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-mongodb.inventory.products_on_hand.Envelope"},"payload":{"before":null,"after":{"product_id":101,"quantity":3},"source":{"version":"2.1.4.Final-redhat-00001","connector":"mongodb","name":"inventory-connector-mongodb","ts_ms":1638985247805,"snapshot":"true","db":"inventory","sequence":null,"table":"products_on_hand","server_id":0,"gtid":null,"file":"mongodb-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。