Confluent Kafkaへの接続

Confluent Schema Registry接続で使用するConfluent Kafkaへの接続を作成し、OCI GoldenGate Big Dataデプロイメントでソースまたはターゲットとして機能します。

開始する前に

次を確認します:

ソース接続の作成

ソースConfluent Kafka接続を作成するには:

  1. OCI GoldenGateの「概要」ページで、「接続」を選択します。

    「はじめに」セクションの「接続の作成」を選択して、ステップ3にスキップすることも可能です。

  2. 「接続」ページで、「接続の作成」を選択します。

  3. 「接続の作成」ページで、次のようにフィールドに入力します:

    1. 「名前」に、接続の名前を入力します。

    2. (オプション)「説明」に、この接続を他の接続から区別するのに役立つ説明を入力します。

    3. (マルチクラウド上のGoldenGateの場合のみ)「サブスクリプション」を選択し、次のフィールドに入力します。

      1. 「コンパートメント」ドロップダウンで、リソース・アンカーが存在するコンパートメントを選択します。

      2. マルチクラウド・パートナ・リージョンを選択します。

      3. 「パートナ可用性ゾーン」を選択します。使用可能なオプションは、選択したマルチクラウド・パートナ・リージョンに基づいて移入されます。

    4. 「コンパートメント」で、接続を作成するコンパートメントを選択します。

    5. 「タイプ」ドロップダウンから、「Confluent Kafka」を選択します。

    6. 「Bootstrap servers:

      1. ブートストラップ・サーバーのホストおよびポート番号を入力します。プライベートIPは、ホスト名がサブネットから解決できない場合、またはSSL/TLSを使用する場合にのみ入力します。

        ノート:プライベートIPを入力すると、OCI GoldenGateはプライベートIPをip-10-0-0-0.ociggsvc.oracle.vcn.comという形式でリライトします。

        ヒント:プライベート・エンドポイントをトラバースできるように、クラスタ内のすべてのノードにFQDNが必要です。

      2. (オプション)別のブートストラップ・サーバを追加するには、「+ブートストラップ・サーバー」を選択します。

    7. 「セキュリティ・プロトコル」で、次の中から選択します:

      • 平文(Plaintext)

      • プレーン・テキストでSASLを実行し、「Username」とパスワードを指定します。

      • SASL over SSL、および必要に応じてユーザー名、パスワード、トラストストア、およびキーストアの値を指定します。

      • SSL、および必要に応じてトラストストアとキーストアの値を指定します。

    8. 拡張オプションの表示を展開します。次のオプションを構成できます。

      • セキュリティ

        • この接続にパスワード・シークレットを使用しない「ボールト・シークレットの使用」の選択を解除します。選択されていない場合:

          • すべての暗号化キー管理をOracleに任せるには、「Oracle管理暗号化キーの使用」を選択します。

          • 「顧客管理暗号化キーの使用」を選択して、OCI Vaultに格納されている特定の暗号化キーを選択し、接続資格証明を暗号化します。

      • ネットワーク接続性

        • 共有エンドポイント: 割り当てられたデプロイメントとエンドポイントを共有します。デプロイメントのイングレスIPからの接続を許可する必要があります。

        • 専用エンドポイント: VCN内の割り当てられたサブネット内の専用エンドポイントを介したネットワーク・トラフィック用。この接続のイングレスIPからの接続を許可する必要があります。

          ノート:

          • 専用接続が7日間未割当てのままの場合、サービスはそれを共有接続に変換します。
          • Oracle GoldenGate接続についてさらに学習します。

      • 設定: Kafkaからキャプチャするには、次のいずれかのデシリアライザまたはコンバータを使用してKafkaコンシューマ・プロパティ・ファイルを作成します:

        • JSONデシリアライザのKafkaコンシューマ・プロパティ:

          key.deserializer=org.apache.kafka.common.serialization.ByteArrayDeserializer
value.deserializer=org.apache.kafka.common.serialization.ByteArrayDeserializer

        • JSONコンバータのKafkaコンシューマ・プロパティ:

          key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter

        • AvroコンバータのKafkaコンシューマ・プロパティ:

          key.converter=io.confluent.connect.avro.AvroConverter
value.converter=io.confluent.connect.avro.AvroConverter

      • セキュリティ属性: Zero Trust Packet Routing (ZPR)を使用して、この接続へのアクセスを制御するセキュリティ属性を追加します。

      • タグ: タグを追加すると、リソースを編成できます。

  4. 「作成」を選択します。

接続が「接続」リストに表示され、そこで接続を選択して詳細を表示できます。Confluentスキーマ・レジストリへの接続も作成し、両方の接続をビッグ・データ・デプロイメントに割り当てます。

ターゲット接続の作成

ターゲットConfluent Kafka接続を作成するには:

  1. OCI GoldenGateの「概要」ページで、「接続」を選択します。

    「はじめに」セクションの「接続の作成」を選択して、ステップ3にスキップすることも可能です。

  2. 「接続」ページで、「接続の作成」を選択します。

  3. 「接続の作成」ページで、次のようにフィールドに入力します:

    1. 「名前」に、接続の名前を入力します。

    2. (オプション)「説明」に、この接続を他の接続から区別するのに役立つ説明を入力します。

    3. (マルチクラウド上のGoldenGateの場合のみ)「サブスクリプション」を選択し、次のフィールドに入力します。

      1. 「コンパートメント」ドロップダウンで、リソース・アンカーが存在するコンパートメントを選択します。

      2. マルチクラウド・パートナ・リージョンを選択します。

      3. 「パートナ可用性ゾーン」を選択します。使用可能なオプションは、選択したマルチクラウド・パートナ・リージョンに基づいて移入されます。

    4. 「コンパートメント」で、接続を作成するコンパートメントを選択します。

    5. 「タイプ」ドロップダウンから、「Confluent Kafka」を選択します。

    6. 「Bootstrap servers:

      1. トラフィック・ルーティング方法を選択します:

      2. ブートストラップ・サーバーのホストおよびポート番号を入力します。プライベートIPは、ホスト名がサブネットから解決できない場合、またはSSL/TLSを使用する場合にのみ入力します。

        ノート:プライベートIPを入力すると、OCI GoldenGateはプライベートIPをip-10-0-0-0.ociggsvc.oracle.vcn.comという形式でリライトします。

        ヒント:プライベート・エンドポイントをトラバースできるように、クラスタ内のすべてのノードにFQDNが必要です。

      3. (オプション)別のブートストラップ・サーバを追加するには、「+ブートストラップ・サーバー」を選択します。

    7. 「セキュリティ・プロトコル」で、次の中から選択します:

      • 平文(Plaintext)

      • プレーン・テキストにSASLを使用し、ユーザー名とパスワードを指定します。

      • SASL over SSL、および必要に応じてユーザー名、パスワード、トラストストア、およびキーストアの値を指定します。

      • SSL、および必要に応じてトラストストアとキーストアの値を指定します。

    8. 拡張オプションの表示を展開します。次のオプションを構成できます。

      • セキュリティ

        • この接続にパスワード・シークレットを使用しない「ボールト・シークレットの使用」の選択を解除します。選択されていない場合:

          • すべての暗号化キー管理をOracleに任せるには、「Oracle管理暗号化キーの使用」を選択します。

          • 「顧客管理暗号化キーの使用」を選択して、OCI Vaultに格納されている特定の暗号化キーを選択し、接続資格証明を暗号化します。

      • ネットワーク接続性

        • 共有エンドポイント: 割り当てられたデプロイメントとエンドポイントを共有します。デプロイメントのイングレスIPからの接続を許可する必要があります。

        • 専用エンドポイント: VCN内の割り当てられたサブネット内の専用エンドポイントを介したネットワーク・トラフィック用。この接続のイングレスIPからの接続を許可する必要があります。

          ノート:

          • 専用接続が7日間未割当てのままの場合、サービスはそれを共有接続に変換します。
          • Oracle GoldenGate接続についてさらに学習します。

      • 設定: KafkaレプリケーションでSnappy圧縮を使用するには、プロデューサ・プロパティをドラッグ・アンド・ドロップまたは選択し、圧縮OCI GoldenGate (Confluent) Kafkaレプリケーションの使用の説明に従ってレプリケーション設定を変更します。

      • セキュリティ属性: Zero Trust Packet Routing (ZPR)を使用して、この接続へのアクセスを制御するセキュリティ属性を追加します。

      • タグ: タグを追加すると、リソースを編成できます。

  4. 「作成」を選択します。

接続が「接続」リストに表示され、そこで接続を選択して詳細を表示できます。Confluentスキーマ・レジストリへの接続も作成し、両方の接続をビッグ・データ・デプロイメントに割り当てます。

プライベート・リンクを使用したConfluent Cloudへの接続の作成

プライベート・リンクを使用すると、仮想ネットワークに存在するプライベート・エンドポイントを介して、サード・パーティ・クラウドで実行されているConfluent Cloudクラスタにアクセスできます。

接続を作成する前に、次のものがあることを確認してください。

前述の手順を使用して接続を作成できますが、ステップ5および6のかわりに次を実行します。

次の手順

Kafka接続エラーのトラブルシューティング

ほとんどの接続の問題では、TimeoutExceptionエラーが発生します。たとえば:

A failure occurred sending a message to Kafka to topic [ggstest] org.apache.kafka.common.errors.TimeoutException: Topic ggstest not  present in metadata after 60000/120000 ms.

Replicatレポート・ファイルでこのメッセージが表示された場合は、次の操作を実行できます。