6.5. Debezium Oracle コネクターのデプロイメント

手順

1. OpenShift クラスターにログインします。

2. コネクターの Debezium KafkaConnect カスタムリソース (CR) を作成するか、既存のリソースを変更します。たとえば、metadata.annotations および spec.build プロパティーを指定する dbz-connect.yaml という名前の KafkaConnect CR を作成します。以下の例は、KafkaConnect カスタムリソースを記述する dbz-connect.yaml ファイルからの抜粋を示しています。
   

   例6.1 Debezium コネクターを含む KafkaConnect カスタムリソースを定義する dbz-connect.yaml ファイル

   次の例では、カスタムリソースは、次のアーティファクトをダウンロードするように設定されています。

   • Debezium Oracle コネクターアーカイブ。
   • サービスレジストリーアーカイブ。Service Registry はオプションのコンポーネントです。コネクターで Avro シリアル化を使用する場合にのみ、Service Registry コンポーネントを追加します。
   • Debezium スクリプト SMT アーカイブと Debezium コネクターで使用する関連言語の依存関係。SMT アーカイブおよび言語の依存関係は任意のコンポーネントです。Debezium コンテンツベースのルーティング SMT または フィルター SMT を使用する場合にのみ、これらのコンポーネントを追加します。
   • Oracle JDBC ドライバー。Oracle データベースに接続するために必要ですが、コネクターアーカイブには含まれていません。

   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-oracle
           artifacts:
             - type: zip 

   6

   
               url: https://maven.repository.redhat.com/ga/io/debezium/debezium-connector-oracle/2.1.4.Final-redhat-00001/debezium-connector-oracle-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
             - type: jar          

   11

   
               url: https://repo1.maven.org/maven2/com/oracle/database/jdbc/ojdbc8/21.1.0.0/ojdbc8-21.1.0.0.jar
   
     bootstrapServers: debezium-kafka-cluster-kafka-bootstrap:9093
   
     ...

   Copy to Clipboard Toggle word wrap

   Expand

   表6.12 Kafka Connect 設定の説明

   項目	説明
   1	strimzi.io/use-connector-resources アノテーションを true に設定して、クラスター Operator が KafkaConnector リソースを使用してこの Kafka Connect クラスター内のコネクターを設定できるようにします。
   2	spec.build 設定は、ビルドイメージの保存場所を指定し、プラグインアーティファクトの場所と共にイメージに追加するプラグインを一覧表示します。
   3	build.output は、新たにビルドされたイメージが保存されるレジストリーを指定します。
   4	イメージ出力の名前およびイメージ名を指定します。output.type の有効な値は、Docker Hub や Quay などのコンテナーレジストリーにプッシュする場合は docker、内部の OpenShift ImageStream にイメージをプッシュする場合は imagestream です。ImageStream を使用するには、ImageStream リソースをクラスターにデプロイする必要があります。KafkaConnect 設定で build.output を指定する方法の詳細は、OpenShift での AMQ Streams の設定の AMQ Streams ビルドスキーマ参照 を参照してください。
   5	plugins 設定は、Kafka Connect イメージに追加するすべてのコネクターを一覧表示します。一覧の各エントリーについて、プラグイン name と、コネクターのビルドに必要なアーティファクトに関する情報を指定します。任意で、各コネクタープラグインに対して、コネクターと使用できる他のコンポーネントを含めることができます。たとえば、Service Registry アーティファクトまたは Debezium スクリプトコンポーネントを追加できます。
   6	artifacts.type の値は、artifacts.url で指定したアーティファクトのファイルタイプを指定します。有効なタイプは zip、tgz、または jar です。Debezium コネクターアーカイブは、.zip ファイル形式で提供されます。JDBC ドライバーファイルは .jar 形式です。type の値は、url フィールドで参照されるファイルのタイプと一致する必要があります。
   7	artifacts.url の値は、コネクターアーティファクトのファイルを格納する Maven リポジトリーなどの HTTP サーバーのアドレスを指定します。Debezium コネクターアーティファクトは Red Hat リポジトリーで入手できます。OpenShift クラスターは指定されたサーバーにアクセスできる必要があります。
   8	(オプション) サービスレジストリーコンポーネントをダウンロードするための アーティファクト type と url を 指定します。デフォルトの JSON コンバーターを使用する代わりに、コネクターが Apache Avro を使用して、Service Registry を使用してイベントのキーと値をシリアル化する場合にのみ、ServiceRegistry アーティファクトを含めます。
   9	(オプション) Debezium コネクターで使用する Debezium スクリプト SMT アーカイブのアーティファクト type と url を指定します。Debezium コンテンツベースルーティング SMT または フィルター SMT を使用する場合にのみ、スクリプト SMT を含めます。スクリプト SMT を使用するには、groovy などの JSR 223 準拠のスクリプト実装もデプロイする必要があります。
   10	(オプション) JSR 223 準拠のスクリプト実装の JAR ファイルのアーティファクト type と url を指定します。これは、Debezium スクリプト SMT で必要です。 重要 AMQ Streams を使用して Kafka Connect イメージにコネクタープラグインを組み込む場合、必要なスクリプト言語コンポーネントごとに、artifacts.url に JAR ファイルの場所を指定し、artifacts.type の値も jar に設定する必要があります。値が無効な場合、実行時にコネクターが失敗します。 スクリプト SMT で Apache Groovy 言語を使用できるようにするために、この例のカスタムリソースは、次のライブラリーの JAR ファイルを取得します。 groovy groovy-jsr223 (スクリプトエージェント) groovy-json (JSON 文字列を解析するためのモジュール) Debezium スクリプト SMT は、GraalVM JavaScript の JSR 223 実装の使用もサポートします。
   11	Maven Central における Oracle JDBC ドライバーの場所を指定します。必要なドライバーは Debezium Oracle コネクターアーカイブには含まれていません。

   Show more

3. 以下のコマンドを入力して、KafkaConnect ビルド仕様を OpenShift クラスターに適用します。

   oc create -f dbz-connect.yaml

   Copy to Clipboard Toggle word wrap

   Streams Operator はカスタムリソースで指定された設定に基づいて、デプロイする Kafka Connect イメージを準備します。
   ビルドが完了すると、Operator はイメージを指定されたレジストリーまたは ImageStream にプッシュし、Kafka Connect クラスターを起動します。設定に一覧表示されているコネクターアーティファクトはクラスターで利用できます。

4. KafkaConnector リソースを作成し、デプロイする各コネクターのインスタンスを定義します。
   たとえば、以下の KafkaConnector CR を作成し、oracle-inventory-connector.yaml として保存します。

   例6.2 Debezium コネクターの KafkaConnector カスタムリソースを定義する oracle-inventory-connector.yaml ファイル

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaConnector
   metadata:
     labels:
       strimzi.io/cluster: debezium-kafka-connect-cluster
     name: inventory-connector-oracle 

   1

   
   spec:
     class: io.debezium.connector.oracle.OracleConnector 

   2

   
     tasksMax: 1  

   3

   
     config:  

   4

   
       schema.history.internal.kafka.bootstrap.servers: debezium-kafka-cluster-kafka-bootstrap.debezium.svc.cluster.local:9092
       schema.history.internal.kafka.topic: schema-changes.inventory
       database.hostname: oracle.debezium-oracle.svc.cluster.local 

   5

   
       database.port: 1521   

   6

   
       database.user: debezium  

   7

   
       database.password: dbz  

   8

   
       database.dbname: mydatabase 

   9

   
       topic.prefix: inventory-connector-oracle 

   10

   
       table.include.list: PUBLIC.INVENTORY  

   11

   
   
       ...

   Copy to Clipboard Toggle word wrap
   Expand

   表6.13 コネクター設定の説明

   項目	説明
   1	Kafka Connect クラスターに登録するコネクターの名前。
   2	コネクタークラスの名前。
   3	同時に動作できるタスクの数。
   4	コネクターの設定。
   5	ホストデータベースインスタンスのアドレス。
   6	データベースインスタンスのポート番号。
   7	Debezium がデータベースへの接続に使用するアカウントの名前。
   8	Debezium がデータベースユーザーアカウントに接続するために使用するパスワード。
   9	変更をキャプチャーするデータベースの名前。
   10	データベースインスタンスまたはクラスターのトピック接頭辞。 指定する名前は、英数字またはアンダースコアのみで設定する必要があります。 トピック接頭辞は、このコネクターから変更イベントを受信する Kafka トピックの接頭辞として使用されるため、名前はクラスターのコネクター間で一意である必要があります。 コネクターを Avro コネクター と統合する場合、この名前空間は関連する Kafka Connect スキーマの名前や、対応する Avro スキーマの名前空間でも使用されます。
   11	コネクターが変更イベントをキャプチャーするテーブルの一覧。

   Show more

5. 以下のコマンドを実行してコネクターリソースを作成します。

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

   Copy to Clipboard Toggle word wrap

   以下に例を示します。

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

   Copy to Clipboard Toggle word wrap

   コネクターは Kafka Connect クラスターに登録され、KafkaConnector CR の spec.config.database.dbname で指定されたデータベースに対して実行を開始します。コネクター Pod の準備ができると、Debezium が実行されます。

手順

1. Kafka Connect の Debezium Oracle コンテナーを作成します。

   1. registry.redhat.io/amq7/amq-streams-kafka-32-rhel8:2.2.0-12 をベースイメージとして使用して、新規の Dockerfile を作成します。例えば、ターミナルウィンドウから、以下のコマンドを入力します。

      cat <<EOF >debezium-container-for-oracle.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-oracle/2.1.4.Final-redhat-00001/debezium-connector-oracle-2.1.4.Final-redhat-00001-plugin.zip \
      && unzip debezium-connector-oracle-2.1.4.Final-redhat-00001-plugin.zip \
      && rm debezium-connector-oracle-2.1.4.Final-redhat-00001-plugin.zip
      RUN cd /opt/kafka/plugins/debezium/ \
      && curl -O https://repo1.maven.org/maven2/com/oracle/ojdbc/ojdbc8/21.1.0.0/ojdbc8-21.1.0.0.jar
      USER 1001
      EOF

      Copy to Clipboard Toggle word wrap
      Expand

      項目	説明
      1	任意のファイル名を指定できます。
      2	Kafka Connect プラグインディレクトリーへのパスを指定します。Kafka Connect のプラグインディレクトリーが別の場所にある場合は、このパスを実際のディレクトリーのパスに置き換えてください。

      Show more

      このコマンドは、現在のディレクトリーに debezium-container-for-oracle.yaml という名前の Docker ファイルを作成します。

   2. 前のステップで作成した debezium-container-for-oracle.yaml Docker ファイルからコンテナーイメージをビルドします。ファイルが含まれるディレクトリーから、ターミナルウィンドウを開き、以下のコマンドのいずれかを入力します。

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

      Copy to Clipboard Toggle word wrap

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

      Copy to Clipboard Toggle word wrap

      上記のコマンドは、debezium-container-for-oracle という名前のコンテナーイメージを構築します。

   3. カスタムイメージを quay.io などのコンテナーレジストリーまたは内部のコンテナーレジストリーにプッシュします。コンテナーレジストリーは、イメージをデプロイする OpenShift インスタンスで利用できる必要があります。以下のいずれかのコマンドを実行します。

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

      Copy to Clipboard Toggle word wrap

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

      Copy to Clipboard Toggle word wrap

   4. 新しい Debezium Oracle KafkaConnect カスタムリソース (CR) を作成します。たとえば、annotations および image プロパティーを指定する dbz-connect.yaml という名前の KafkaConnect CR を作成します。以下の例は、KafkaConnect カスタムリソースを記述する dbz-connect.yaml ファイルからの抜粋を示しています。
      

      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-oracle 

      2

      
      
        ...

      Copy to Clipboard Toggle word wrap
      Expand

      項目	説明
      1	KafkaConnector リソースはこの Kafka Connect クラスターでコネクターを設定するために使用されることを、metadata.annotations は Cluster Operator に示します。
      2	spec.image は Debezium コネクターを実行するために作成したイメージの名前を指定します。設定された場合、このプロパティーによって Cluster Operator の STRIMZI_DEFAULT_KAFKA_CONNECT_IMAGE 変数がオーバーライドされます。

      Show more

   5. 以下のコマンドを入力して、KafkaConnect CR を OpenShift Kafka Connect 環境に適用します。

      oc create -f dbz-connect.yaml

      Copy to Clipboard Toggle word wrap

      このコマンドは、Debezium コネクターを実行するために作成したイメージの名前を指定する Kafka Connect インスタンスを追加します。

2. Debezium Oracle コネクターインスタンスを設定する KafkaConnector カスタムリソースを作成します。

   通常、コネクターに使用できる設定プロパティーを使用して、.yaml ファイルに Debezium Db2 コネクターを設定します。コネクター設定は、Debezium に対して、スキーマおよびテーブルのサブセットにイベントを生成するよう指示する可能性があり、または機密性の高い、大きすぎる、または不必要な指定のコラムで Debezium が値を無視、マスク、または切り捨てするようにプロパティーを設定する可能性もあります。

   以下の例では、ポート 1521 で Oracle ホスト IP アドレスに接続する Debezium コネクターを設定します。このホストには ORCLCDB という名前のデータベースがあり、server1 はサーバーの論理名です。

   Oracle inventory-connector.yaml

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaConnector
   metadata:
     name: inventory-connector-oracle 

   1

   
     labels:
       strimzi.io/cluster: my-connect-cluster
     annotations:
       strimzi.io/use-connector-resources: 'true'
   spec:
     class: io.debezium.connector.oracle.OracleConnector 

   2

   
     config:
       database.hostname: <oracle_ip_address> 

   3

   
       database.port: 1521 

   4

   
       database.user: c##dbzuser 

   5

   
       database.password: dbz 

   6

   
       database.dbname: ORCLCDB 

   7

   
       database.pdb.name : ORCLPDB1, 

   8

   
       topic.prefix: inventory-connector-oracle 

   9

   
       schema.history.internal.kafka.bootstrap.servers: kafka:9092 

   10

   
       schema.history.internal.kafka.topic: schema-changes.inventory 

   11

   Copy to Clipboard Toggle word wrap

   Expand

   表6.14 コネクター設定の説明

   項目	説明
   1	Kafka Connect サービスに登録する場合のコネクターの名前。
   2	この Oracle コネクタークラスの名前。
   3	Oracle インスタンスのアドレス。
   4	Oracle インスタンスのポート番号。
   5	コネクターのユーザーの作成 で指定した Oracle ユーザーの名前。
   6	コネクターのユーザーの作成 で指定した Oracle ユーザーのパスワード。
   7	変更をキャプチャーするデータベースの名前。
   8	コネクターが変更をキャプチャーする Oracle のプラグ可能なデータベースの名前。コンテナーデータベース (CDB) のインストールのみで使用されます。
   9	topic prefix は、コネクターが変更をキャプチャーする Oracle データベースサーバーの名前空間を識別して提供します。
   10	DDL ステートメントをデータベーススキーマの履歴トピックに書き込み、復元するためにコネクターによって使用される Kafka ブローカーのリスト。
   11	コネクターが DDL ステートメントを書き、復元するデータベーススキーマの履歴トピックの名前。このトピックは内部使用のみを目的としており、コンシューマーが使用しないようにしてください。

   Show more

3. Kafka Connect でコネクターインスタンスを作成します。たとえば、KafkaConnector リソースを inventory-connector.yaml ファイルに保存した場合は、以下のコマンドを実行します。

   oc apply -f inventory-connector.yaml

   Copy to Clipboard Toggle word wrap

   上記のコマンドは inventory-connector を登録し、コネクターは KafkaConnector CR に定義されている server1 データベースに対して実行を開始します。

手順

1. 以下の方法のいずれかを使用して KafkaConnector リソースのステータスを確認します。

   • OpenShift Container Platform Web コンソールから以下を実行します。

     1. Home Search に移動します。
     2. Search ページで Resources をクリックし、Select Resource ボックスを開き、KafkaConnector を入力します。
     3. KafkaConnectors リストから、チェックするコネクターの名前をクリックします (例: inventory-connector-oracle)。
     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-oracle -n debezium

        Copy to Clipboard Toggle word wrap

        このコマンドは、以下の出力のようなステータス情報を返します。

        例6.3 KafkaConnector リソースのステータス

        Name:         inventory-connector-oracle
        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-oracle
            Tasks:
              Id:               0
              State:            RUNNING
              worker_id:        10.131.1.124:8083
            Type:               source
          Observed Generation:  1
          Tasks Max:            1
          Topics:
            inventory-connector-oracle.inventory
            inventory-connector-oracle.inventory.addresses
            inventory-connector-oracle.inventory.customers
            inventory-connector-oracle.inventory.geom
            inventory-connector-oracle.inventory.orders
            inventory-connector-oracle.inventory.products
            inventory-connector-oracle.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 リストから確認するトピックの名前をクリックします (例: inventory-connector-oracle.inventory.orders---ac5e98ac6a5d91e04d8ec0dc9078a1ece439081d)。
     4. Conditions セクションで、Type および Status 列の値が Ready および True に設定されていることを確認します。

   • ターミナルウィンドウから以下を実行します。

     1. 以下のコマンドを入力します。

        oc get kafkatopics

        Copy to Clipboard Toggle word wrap

        このコマンドは、以下の出力のようなステータス情報を返します。

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

        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 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-oracle.inventory.products_on_hand

   Copy to Clipboard Toggle word wrap

   トピック名を指定する形式は、手順 1 で返された oc describe コマンドと同じです (例: inventory-connector-oracle.inventory.addresses)。

   トピックの各イベントについて、このコマンドは、以下の出力のような情報を返します。

   例6.5 Debezium 変更イベントの内容

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

   Copy to Clipboard Toggle word wrap

   上記の例では、payload 値は、コネクタースナップショットがテーブル inventory.products_on_hand から 読み込み (op" ="r") イベントを生成したことを示しています。product_id レコードの before 状態は null であり、レコードに以前の値が存在しないことを示します。"after" 状態が product_id 101 で項目の quantity を 3 で示しています。