11.2. Avro シリアライゼーションを使用する Debezium コネクターの設定

1. Red Hat build of Apicurio Registry インスタンスの インストールとデプロイ の手順に従って、Red Hat build of Apicurio Registry インスタンスをデプロイします。
2. Debezium Service Registry Kafka Connect の zip ファイルをダウンロードして Debezium コネクターのディレクトリーにデプロイメントし、Avro コンバーターをインストールする。

3. 以下のように設定プロパティーを設定して、Avro シリアライゼーションを使用するように Debezium コネクターインスタンスを設定する。

   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

   Copy to Clipboard

手順

1. Apicurio Registry のインスタンスをデプロイします。OpenShift での Red Hat build of Apicurio Registry の インストールとデプロイ を参照してください。

   • Apicurio 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
       ├── ...

   Copy to Clipboard

3. Avro シリアライゼーションを使用するように設定する Debezium コネクターが含まれるディレクトリーに Avro コンバーターを追加します。

   1. Red Hat build of Debezium ダウンロードサイト に移動し、Apicurio Registry Kafka Connect zip ファイルをダウンロードします。
   2. 目的の Debezium コネクターディレクトリーにアーカイブをデプロイメントします。

   複数のタイプの Debezium コネクターを Avro シリアライゼーションを使用するように設定するには、該当するそれぞれのコネクタータイプのディレクトリーにアーカイブをデプロイメントします。それぞれのディレクトリーにアーカイブを抽出するとファイルが重複しますが、これにより依存関係の競合が生じる可能性がなくなります。

4. Avro コンバーターを使用するように設定する Debezium コネクターを実行するためのカスタムイメージを作成して公開します。

   1. registry.redhat.io/amq-streams-kafka-35-rhel8:2.5.0 をベースイメージとして使用して、新しい Dockerfile を作成します。以下の例の my-plugins を、実際のプラグインディレクトリーの名前に置き換えてください。

      FROM registry.redhat.io/amq-streams-kafka-35-rhel8:2.5.0
      USER root:root
      COPY ./my-plugins/ /opt/kafka/plugins/
      USER 1001

      Copy to Clipboard

      Kafka Connect は、コネクターの実行を開始する前に、/opt/kafka/plugins ディレクトリーにあるサードパーティープラグインをロードします。

   2. docker コンテナーイメージをビルドします。例えば、前のステップで作成した docker ファイルを debezium-container-with-avro として保存した場合、以下のコマンドを実行します。

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

   3. カスタムイメージをコンテナーレジストリーにプッシュします。例を以下に示します。

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

   4. 新しいコンテナーイメージを示します。次のいずれかを行います。

      • KafkaConnect カスタムリソースの KafkaConnect.spec.image プロパティーを編集します。このプロパティーが設定されていると、クラスターオペレータの STRIMZI_DEFAULT_KAFKA_CONNECT_IMAGE 変数がオーバーライドされます。以下に例を示します。

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

        Copy to Clipboard
      • install/cluster-operator/050-Deployment-strimzi-cluster-operator.yaml ファイルの STRIMZI_DEFAULT_KAFKA_CONNECT_IMAGE 変数を編集し、新しいコンテナーイメージを示すようにした後、Cluster Operator を再インストールします。このファイルを編集する場合は、これを OpenShift クラスターに適用する必要があります。

5. Avro コンバーターを使用するように設定されたそれぞれの Debezium コネクターをデプロイします。それぞれの Debezium コネクターについて、以下の設定を行います。

   1. Debezium コネクターインスタンスを作成します。次の inventory-connector.yaml ファイルの例では、Avro コンバーターを使用するように設定された My SQL コネクターインスタンスを定義する Kafka Connector カスタムリソースを作成しています。

      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

      Copy to Clipboard

   2. コネクターインスタンスを適用します。以下に例を示します。

      oc apply -f inventory-connector.yaml

      これにより inventory-connector が登録され、コネクターが inventory データベースに対して実行されるようになります。

6. コネクターが作成され、指定されたデータベース内の変更の追跡を開始したことを確認します。例えば inventory-connector が起動したときの Kafka Connect のログ出力を見ることで、コネクターのインスタンスを確認することができます。

   1. Kafka Connect のログ出力を表示します。

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

      Copy to Clipboard

   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]
      ...

      Copy to Clipboard

      スナップショットは、複数のステップを経て作成されます。

      ...
      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]
      ...

      Copy to Clipboard

      スナップショットの作成が完了した後、Debezium は (例として) inventory データベースの 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]
      ...

      Copy to Clipboard