3.3. 验证连接器部署

流程

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

   • 在 OpenShift Container Platform Web 控制台中：

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

   • 在一个终端窗口中：

     1. 使用以下命令：

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

        Copy to Clipboard Toggle word wrap

        例如，

        oc describe KafkaConnector inventory-connector -n debezium

        Copy to Clipboard Toggle word wrap

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

        例 3.3. KafkaConnector 资源状态

        Name:         inventory-connector
        Namespace:    debezium
        Labels:       strimzi.io/cluster=my-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
            Tasks:
              Id:               0
              State:            RUNNING
              worker_id:        10.131.1.124:8083
            Type:               source
          Observed Generation:  1
          Tasks Max:            1
          Topics:
            dbserver1
            dbserver1.inventory.addresses
            dbserver1.inventory.customers
            dbserver1.inventory.geom
            dbserver1.inventory.orders
            dbserver1.inventory.products
            dbserver1.inventory.products_on_hand
        Events:  <none>

        Copy to Clipboard Toggle word wrap

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

   • 通过 OpenShift Container Platform Web 控制台。

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

   • 在一个终端窗口中：

     1. 使用以下命令：

        oc get kafkatopics

        Copy to Clipboard Toggle word wrap

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

        例 3.4. KafkaTopic 资源状态

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

        Copy to Clipboard Toggle word wrap

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>

   Copy to Clipboard Toggle word wrap

   例如，

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

   Copy to Clipboard Toggle word wrap

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

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

   例 3.5. Debezium 更改事件的内容

   {"schema":{"type":"struct","fields":[{"type":"int32","optional":false,"field":"product_id"}],"optional":false,"name":"dbserver1.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":"dbserver1.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":"dbserver1.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":"dbserver1.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":"dbserver1","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}}

   Copy to Clipboard Toggle word wrap

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