11.2. 配置 Debezium 连接器以使用 Avro 序列化

1. 按照在 OpenShift 上安装和部署 Service Registry 中的说明部署 Service Registry 实例。
2. 通过下载 Debezium Service Registry Kafka Connect zip 文件并将其提取到 Debezium 连接器的目录中来安装 Avro converter。

3. 通过设置配置属性将 Debezium 连接器实例配置为使用 Avro serialization，如下所示：

   key.converter=io.apicurio.registry.utils.converter.AvroConverter
   key.converter.apicurio.registry.url=http://apicurio:8080/apis/registry/v2
   key.converter.apicurio.registry.auto-register=true
   key.converter.apicurio.registry.find-latest=true
   value.converter=io.apicurio.registry.utils.converter.AvroConverter
   value.converter.apicurio.registry.url=http://apicurio:8080/apis/registry/v2
   value.converter.apicurio.registry.auto-register=true
   value.converter.apicurio.registry.find-latest=true
   schema.name.adjustment.mode=avro

流程

1. 部署 Service Registry 实例。请参阅 在 OpenShift 上安装和部署 Service Registry，其提供了以下说明：

   • 安装 Service Registry
   • 安装 AMQ Streams
   • 设置 AMQ Streams 存储

2. 提取 Debezium 连接器存档，以便为连接器插件创建目录结构。如果您下载并提取多个 Debezium 连接器的存档，生成的目录结构类似以下示例：

   tree ./my-plugins/
   ./my-plugins/
   ├── debezium-connector-mongodb
   |   ├── ...
   ├── debezium-connector-mysql
   │   ├── ...
   ├── debezium-connector-postgres
   │   ├── ...
   └── debezium-connector-sqlserver
       ├── ...

3. 将 Avro converter 添加到包含您要配置为使用 Avro 序列化的 Debezium 连接器的目录中：

   1. 进入 Red Hat Integration 下载站点 并下载 Service Registry Kafka Connect zip 文件。
   2. 将存档提取到所需的 Debezium 连接器目录中。

   要将多种 Debezium 连接器配置为使用 Avro 序列化，请将存档提取到每个相关连接器类型的目录中。虽然将存档提取到每个目录中会复制文件，您可以删除可能冲突的依赖项。

4. 创建并发布自定义镜像，以运行配置为使用 Avro converter 的 Debezium 连接器：

   1. 使用 registry.redhat.io/amq7/amq-streams-kafka-32-rhel8:2.2.0-12 作为基础镜像来创建新的 Dockerfile。在以下示例中，将 my-plugins 替换为插件目录的名称：

      FROM registry.redhat.io/amq7/amq-streams-kafka-32-rhel8:2.2.0-12
      USER root:root
      COPY ./my-plugins/ /opt/kafka/plugins/
      USER 1001

      在 Kafka Connect 开始运行连接器前，Kafka Connect 会加载 /opt/kafka/plugins 目录中的任何第三方插件。

   2. 构建 docker 容器镜像。例如，如果您将您在上一步中创建的 docker 文件保存为 debezium-container-with-avro，则可运行以下命令：

      docker build -t debezium-container-with-avro:latest

   3. 将自定义镜像推送到容器 registry，例如：

      docker push <myregistry.io>/debezium-container-with-avro:latest

   4. 指向新容器镜像。执行以下操作之一：

      • 编辑 KafkaConnect 自定义资源的 KafkaConnect.spec.image 属性。如果设置，此属性会覆盖 Cluster Operator 中的 STRIMZI_DEFAULT_KAFKA_CONNECT_IMAGE 变量。例如：

        apiVersion: kafka.strimzi.io/v1beta2
        kind: KafkaConnect
        metadata:
          name: my-connect-cluster
        spec:
          #...
          image: debezium-container-with-avro

      • 在 install/cluster-operator/050-Deployment-strimzi-cluster-operator.yaml 文件中，编辑 STRIMZI_DEFAULT_KAFKA_CONNECT_IMAGE 变量以指向新容器镜像并重新安装 Cluster Operator。如果编辑此文件，则需要将其应用到 OpenShift 集群。

5. 部署配置为使用 Avro converter 的每个 Debezium 连接器。对于每个 Debezium 连接器：

   1. 创建 Debezium 连接器实例。以下 inventory-connector.yaml 文件示例会创建一个 KafkaConnector 自定义资源，用于定义配置为使用 Avro converter 的 MySQL 连接器实例：

      apiVersion: kafka.strimzi.io/v1beta1
      kind: KafkaConnector
      metadata:
        name: inventory-connector
        labels:
          strimzi.io/cluster: my-connect-cluster
      spec:
        class: io.debezium.connector.mysql.MySqlConnector
        tasksMax: 1
        config:
          database.hostname: mysql
          database.port: 3306
          database.user: debezium
          database.password: dbz
          database.server.id: 184054
          topic.prefix: dbserver1
          database.include.list: inventory
          schema.history.internal.kafka.bootstrap.servers: my-cluster-kafka-bootstrap:9092
          schema.history.internal.kafka.topic: schema-changes.inventory
          schema.name.adjustment.mode: avro
          key.converter: io.apicurio.registry.utils.converter.AvroConverter
          key.converter.apicurio.registry.url: http://apicurio:8080/api
          key.converter.apicurio.registry.global-id: io.apicurio.registry.utils.serde.strategy.GetOrCreateIdStrategy
          value.converter: io.apicurio.registry.utils.converter.AvroConverter
          value.converter.apicurio.registry.url: http://apicurio:8080/api
          value.converter.apicurio.registry.global-id: io.apicurio.registry.utils.serde.strategy.GetOrCreateIdStrategy

   2. 应用连接器实例，例如：

      oc apply -f inventory-connector.yaml

      这会注册 inventory-connector，连接器开始针对 清单 数据库运行。

6. 验证连接器是否已创建，并已开始跟踪指定数据库中的更改。您可以通过观察 Kafka Connect 日志输出来验证连接器实例，例如 inventory-connector 启动。

   1. 显示 Kafka Connect 日志输出：

      oc logs $(oc get pods -o name -l strimzi.io/name=my-connect-cluster-connect)

   2. 检查日志输出，以验证是否已执行初始快照。您应该看到类似以下行的内容：

      ...
      2020-02-21 17:57:30,801 INFO Starting snapshot for jdbc:mysql://mysql:3306/?useInformationSchema=true&nullCatalogMeansCurrent=false&useSSL=false&useUnicode=true&characterEncoding=UTF-8&characterSetResults=UTF-8&zeroDateTimeBehavior=CONVERT_TO_NULL&connectTimeout=30000 with user 'debezium' with locking mode 'minimal' (io.debezium.connector.mysql.SnapshotReader) [debezium-mysqlconnector-dbserver1-snapshot]
      2020-02-21 17:57:30,805 INFO Snapshot is using user 'debezium' with these MySQL grants: (io.debezium.connector.mysql.SnapshotReader) [debezium-mysqlconnector-dbserver1-snapshot]
      ...

      创建快照涉及几个步骤：

      ...
      2020-02-21 17:57:30,822 INFO Step 0: disabling autocommit, enabling repeatable read transactions, and setting lock wait timeout to 10 (io.debezium.connector.mysql.SnapshotReader) [debezium-mysqlconnector-dbserver1-snapshot]
      2020-02-21 17:57:30,836 INFO Step 1: flush and obtain global read lock to prevent writes to database (io.debezium.connector.mysql.SnapshotReader) [debezium-mysqlconnector-dbserver1-snapshot]
      2020-02-21 17:57:30,839 INFO Step 2: start transaction with consistent snapshot (io.debezium.connector.mysql.SnapshotReader) [debezium-mysqlconnector-dbserver1-snapshot]
      2020-02-21 17:57:30,840 INFO Step 3: read binlog position of MySQL primary server (io.debezium.connector.mysql.SnapshotReader) [debezium-mysqlconnector-dbserver1-snapshot]
      2020-02-21 17:57:30,843 INFO 	 using binlog 'mysql-bin.000003' at position '154' and gtid '' (io.debezium.connector.mysql.SnapshotReader) [debezium-mysqlconnector-dbserver1-snapshot]
      ...
      2020-02-21 17:57:34,423 INFO Step 9: committing transaction (io.debezium.connector.mysql.SnapshotReader) [debezium-mysqlconnector-dbserver1-snapshot]
      2020-02-21 17:57:34,424 INFO Completed snapshot in 00:00:03.632 (io.debezium.connector.mysql.SnapshotReader) [debezium-mysqlconnector-dbserver1-snapshot]
      ...

      完成快照后，Debebe 开始跟踪更改，例如，清单 数据库的 binlog 以更改事件：

      ...
      2020-02-21 17:57:35,584 INFO Transitioning from the snapshot reader to the binlog reader (io.debezium.connector.mysql.ChainedReader) [task-thread-inventory-connector-0]
      2020-02-21 17:57:35,613 INFO Creating thread debezium-mysqlconnector-dbserver1-binlog-client (io.debezium.util.Threads) [task-thread-inventory-connector-0]
      2020-02-21 17:57:35,630 INFO Creating thread debezium-mysqlconnector-dbserver1-binlog-client (io.debezium.util.Threads) [blc-mysql:3306]
      Feb 21, 2020 5:57:35 PM com.github.shyiko.mysql.binlog.BinaryLogClient connect
      INFO: Connected to mysql:3306 at mysql-bin.000003/154 (sid:184054, cid:5)
      2020-02-21 17:57:35,775 INFO Connected to MySQL binlog at mysql:3306, starting at binlog file 'mysql-bin.000003', pos=154, skipping 0 events plus 0 rows (io.debezium.connector.mysql.BinlogReader) [blc-mysql:3306]
      ...