※ 本記事は、Sunil Vernekarによる”Using Custom Properties for Confluent Kafka in OCI Goldengate Service Replication Setup“を翻訳したものです。

2023年9月26日

皆さん、ようこそ…

このブログでは、カスタム・プロパティを使用してConfluent KafkaにデータをレプリケートするようにOCI Goldengateサービスを設定します。

OCI GoldengateのConfluent KafkaへのGGレプリケーションを設定する際、ほとんどのプロパティが自動生成されるため、顧客プロパティを使用する必要がある場合は、それらのカスタム・プロパティをアップロードする必要があります。

このブログでは、OCI GG Serviceで、Confluent Kafkaのカスタム・プロパティをアップロードする方法について説明します。

 

アーキテクチャ:

Arch-1

 

ユースケース:

この設定では、スキーマの変更を制御するため、スキーマ・レジストリへの読取り専用アクセス権のみを提供する必要があります。

デフォルトでは、ソース表でDDLが変更された場合、OCI GGはスキーマ・レジストリを新しいスキーマで更新します。

OCI GoldenGateは、デフォルトでAvroおよびJSONコンバータをサポートしていますが、key.convertorにはStringConvertorを、value.convertorにはAvroConvertorを使用する必要がありました。

そのため、カスタム・プロデューサ・プロパティ・ファイルを作成し、それをOCI GoldenGateにアップロードしてGoldengate replicatプロセスで使用する必要があります。

OCI GGサービスの接続セクションでカスタム・プロパティを更新できます。

 

設定の手順は次のとおりです:

1. GGがレプリケートされるすべての表の初期スキーマを作成するためには、最初に書込みアクセス権が必要です。そのため、レプリケーションに新しい表を追加するたびに、GGが初期スキーマを送信するためのスキーマ・レジストリへの書込みアクセス権を提供する必要があります。

 

2. Confluent Kafkaにレプリケートしながら、2つの接続を作成します。これらの接続がすでに作成されていると仮定します

  • Kafka接続
  • スキーマ・レジストリ接続

次のブログに従って、Kafkaプラットフォームへのレプリケーションを設定できます。

KafkaプラットフォームでのOracle Cloud Infrastructure GoldenGateの使用

スキーマ・レジストリへの書込みアクセス権がある間は、次のカスタム・プロパティを使用する必要があります。そのため、次のプロパティでファイルを作成し、Kafka接続にアップロードします

key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=io.confluent.connect.avro.AvroConverter
value.converter.auto.register.schemas=true
value.converter.use.latest.version=false

または、新しいkafka接続の作成時にアップロードできます。

カスタム・プロパティをアップロードするオプションは、次のとおりです

Show Advanced Options à Update additional properties à Producer properties

Arch-2

Arch-3

Arch-4

 

3.  レプリケーションですべての表のスキーマが作成されたら、replicatを停止して、GGユーザーのスキーマ・レジストリから書込みアクセス権を削除できます。

     次に、kafka接続を編集し、次のプロパティを使用してファイルをアップロードします。

    Show Advanced Options à Update additional properties à Producer properties

key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=io.confluent.connect.avro.AvroConverter
value.converter.auto.register.schemas=false
value.converter.use.latest.version=true

 

4. 上記の両方の手順中に、以下のreplicatのプロパティ・ファイルが作成されます

# Properties file for Replicat RConf
#Kafka Connect Handler Template
gg.handlerlist=kafkaconnect
gg.handler.kafkaconnect.type=kafkaconnect
gg.handler.kafkaconnect.connectionId=ocid1.goldengateconnection.oc1.ca-montreal-1.amaaaaaakgimj3aapgarjbnrh26s46i6khcdddehu5j6t6pjehx5yxgm2v2q
gg.handler.kafkaconnect.mode=op
#TODO: Set the template for resolving the topic name.
gg.handler.kafkaconnect.topicMappingTemplate=xxxxxxxxx
gg.handler.kafkaconnect.keyMappingTemplate=${primaryKeys}
gg.handler.kafkaconnect.messageFormatting=row
gg.handler.kafkaconnect.metaColumnsTemplate=${objectname[table]},${optype[op_type]},${timestamp[op_ts]},${currenttimestamp[current_ts]},${position[pos]}
gg.handler.kafkaconnect.schemaRegistryConnectionId=ocid1.goldengateconnection.oc1.ca-montreal-1.amaaaaaakgimj3aa46suka7gy54ttzxyopcuyxhwaj5hdn4uuhmiwhl67txq
#gg.handler.kafkaconnect.converter=avro
#TODO: Set the location of the Kafka client libraries
gg.classpath=$THIRD_PARTY_DIR/kafka/*
jvm.bootoptions=-Xmx512m -Xms32m

唯一の変更は、以下のプロパティをコメント・アウトすることです –> #gg.handler.kafkaconnect.converter=avro

 

まとめ:

このブログでは、データ・レプリケーションを構成し、スキーマ・レジストリへの読取り専用アクセス権を持つConfluent Kafkaにカスタム・プロパティを使用する方法を学習しました。