2.3. 验证连接器部署

如果连接器正确启动且没有错误，它会为每个连接器配置为捕获的表创建一个主题。下游应用程序可以订阅这些主题，以检索源数据库中发生的信息事件。

要验证连接器是否正在运行，您可以从 OpenShift Container Platform Web 控制台或 OpenShift CLI 工具(oc)执行以下操作：

验证连接器状态。
验证连接器是否生成主题。
验证主题是否填充了用于读取操作的事件("op":"r")中连接器在各个表的初始快照期间生成的事件。

先决条件

在 OpenShift 中将一个 Debezium 连接器部署到 AMQ Streams 中。
已安装 OpenShift oc CLI 客户端。
访问 OpenShift Container Platform web 控制台。

流程

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

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

在终端窗口中：

使用以下命令：

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

例如，

oc describe KafkaConnector inventory-connector -n debezium

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

例 2.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:
    inventory_connector
    inventory_connector.inventory.addresses
    inventory_connector.inventory.customers
    inventory_connector.inventory.geom
    inventory_connector.inventory.orders
    inventory_connector.inventory.products
    inventory_connector.inventory.products_on_hand
Events:  <none>

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

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

在终端窗口中：

使用以下命令：

oc get kafkatopics

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

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

检查主题内容。

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

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 my-cluster-kafka-0 -- /opt/kafka/bin/kafka-console-consumer.sh \
>     --bootstrap-server localhost:9092 \
>     --from-beginning \
>     --property print.key=true \
>     --topic=inventory_connector.inventory.products_on_hand

指定主题名称的格式与步骤 1 中介绍的 oc describe 命令相同，如 inventory_connector.inventory.addresses。

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

例 2.5. Debezium 更改事件的内容

{"schema":{"type":"struct","fields":[{"type":"int32","optional":false,"field":"product_id"}],"optional":false,"name":"inventory_connector.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.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.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.inventory.products_on_hand.Envelope"},"payload":{"before":null,"after":{"product_id":101,"quantity":3},"source":{"version":"1.9.5.Final-redhat-00001","connector":"mysql","name":"inventory_connector","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}}

在前面的示例中，有效负载 值显示，连接器快照从表 清单中生成读("op" ="r")事件。product_on_hand.product_id 记录的 "前面" 状态为 null，表示记录不存在之前的值。"after" 状态显示带有 product_id 101 的项目 的数量 3。

2.3. 验证连接器部署

学习

尝试、购买和销售

社区

关于红帽文档

让开源更具包容性

關於紅帽

Red Hat legal and privacy links

Red Hat legal and privacy links