Confluent Kafkaへの接続

OCI GoldenGateビッグ・データ・デプロイメントでソースまたはターゲットとして機能する、Confluent Schema Registry接続で使用するConfluent Kafkaへの接続の作成について学習します。

関連項目

開始する前に

接続を作成する前に、次のことを確認します。

ソース接続の作成

ソースConfluent Kafka接続を作成するには:
  1. OCI GoldenGateの「概要」ページで、「接続」をクリックします。
    「はじめに」セクションの「接続の作成」をクリックして、ステップ3にスキップすることもできます。
  2. 「接続」ページで、「接続の作成」をクリックします。
  3. 「接続の作成」ページで、次のようにフィールドに入力します。
    1. 「名前」に、接続の名前を入力します。
    2. 「説明」に、この接続を他の接続から区別するのに役立つ説明を入力します。
    3. 「コンパートメント」で、接続を作成するコンパートメントを選択します。
    4. 「タイプ」ドロップダウンから、「Confluent Kafka」を選択します。
    5. 「Bootstrap Servers」の下:
      1. ブートストラップ・サーバーのホストおよびポート番号を入力します。サブネットからホスト名を解決できない場合、またはSSL/TLSを使用する場合のみ、プライベートIPを入力します。

        ノート:

        プライベートIPを入力すると、OCI GoldenGateによって、プライベートIPがip-10-0-0-0.ociggsvc.oracle.vcn.comの形式でリライトされます。

        ヒント :

        クラスタ内のすべてのノードに、プライベート・エンドポイントでのトラバースを許可するFQDNが必要です。
      2. (オプション)別のブートストラップ・サーバーを追加するには、「+ブートストラップ・サーバー」をクリックします。
    6. 「セキュリティ・プロトコル」で、次から選択します:
      • 平文(Plaintext)
      • SASL over plaintext、およびユーザー名およびパスワードを指定します。
      • SASL over SSL。必要に応じて、「ユーザー名」、「パスワード」、「トラストストア」および「キーストア」の値を指定します。
      • SSLを使用し、必要に応じてトラストストアおよびキーストアの値を指定します。
    7. 「拡張オプションの表示」を拡張します。構成できるオプションは、次のとおりです。
      • セキュリティ
        • この接続にパスワード・シークレットを使用しない「ボールト・シークレットの使用」の選択を解除します。選択されていない場合:
          • すべての暗号化キー管理を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
      • タグ: タグを追加してリソースを編成します。
  4. 「作成」をクリックします。
接続が「接続」リストに表示され、そこで接続を選択して詳細を表示できます。Confluent Schema Registryへの接続も作成し、両方の接続をビッグ・データ・デプロイメントに割り当てます。

ターゲット接続の作成

ターゲットのConfluent Kafka接続を作成するには:
  1. OCI GoldenGateの「概要」ページで、「接続」をクリックします。
    「はじめに」セクションの「接続の作成」をクリックして、ステップ3にスキップすることもできます。
  2. 「接続」ページで、「接続の作成」をクリックします。
  3. 「接続の作成」ページで、次のようにフィールドに入力します。
    1. 「名前」に、接続の名前を入力します。
    2. 「説明」に、この接続を他の接続から区別するのに役立つ説明を入力します。
    3. 「コンパートメント」で、接続を作成するコンパートメントを選択します。
    4. 「タイプ」ドロップダウンから、「Confluent Kafka」を選択します。
    5. 「Bootstrap Servers」の下:
      1. 「トラフィック・ルーティング方法」を選択します。
      2. ブートストラップ・サーバーのホストおよびポート番号を入力します。サブネットからホスト名を解決できない場合、またはSSL/TLSを使用する場合のみ、プライベートIPを入力します。

        ノート:

        プライベートIPを入力すると、OCI GoldenGateによって、プライベートIPがip-10-0-0-0.ociggsvc.oracle.vcn.comの形式でリライトされます。

        ヒント :

        クラスタ内のすべてのノードに、プライベート・エンドポイントでのトラバースを許可するFQDNが必要です。
      3. (オプション)別のブートストラップ・サーバーを追加するには、「+ブートストラップ・サーバー」をクリックします。
    6. 「セキュリティ・プロトコル」で、次から選択します:
      • 平文(Plaintext)
      • SASL over plaintext、およびユーザー名およびパスワードを指定します。
      • SASL over SSL。必要に応じて、「ユーザー名」、「パスワード」、「トラストストア」および「キーストア」の値を指定します。
      • SSLを使用し、必要に応じてトラストストアおよびキーストアの値を指定します。
    7. 「拡張オプションの表示」を拡張します。構成できるオプションは、次のとおりです。
      • セキュリティ
        • この接続にパスワード・シークレットを使用しない「ボールト・シークレットの使用」の選択を解除します。選択されていない場合:
          • すべての暗号化キー管理をOracleに残すには、「Oracle管理暗号化キーを使用」を選択します。
          • 「顧客管理暗号化キーの使用」を選択して、OCI Vaultに格納されている特定の暗号化キーを選択し、接続資格証明を暗号化します。
      • ネットワーク接続性
        • 共有エンドポイント: 割り当てられたデプロイメントとエンドポイントを共有します。デプロイメントのイングレスIPからの接続を許可する必要があります。
        • 専用エンドポイント: VCN内の割り当てられたサブネット内の専用エンドポイントを介したネットワーク・トラフィック用。この接続のイングレスIPからの接続を許可する必要があります。

          ノート:

          • 専用接続が7日間未割当てのままの場合、サービスはそれを共有接続に変換します。
          • Oracle GoldenGate接続についてさらに学習します。
      • 設定KafkaレプリケーションでSnappy圧縮を使用するには、プロデューサ・プロパティをドラッグ・アンド・ドロップするか選択し、圧縮OCI GoldenGate (Confluent) Kafkaレプリケーションの使用の説明に従ってレプリケーション設定を変更します。
      • タグ: タグを追加してリソースを編成します。
  4. 「作成」をクリックします。
接続が「接続」リストに表示され、そこで接続を選択して詳細を表示できます。Confluent Schema Registryへの接続も作成し、両方の接続をビッグ・データ・デプロイメントに割り当てます。

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

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

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

  • Oracle Cloud Infrastructure(OCI)とターゲット・サード・パーティ・クラウド間のプライベート・ネットワーク接続を作成します。
  • Confluent Cloudでプライベート・リンクのネットワーク構成を追加する場合は、「プライベートDNS解決」を選択してください。
  • Confluent CloudおよびOCIで構成したサード・パーティ・クラウドで、DNSゾーンを構成し、DNSレコードを設定します。OCIでは、VCNのプライベート・ビュー内にゾーンを作成できます。ゾーン内で、必要なDNSレコードを追加できます。

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

  • ステップ3eの場合:
    • ブートストラップ・サーバーのホストおよびポートの詳細を指定します。
    • 複数のブートストラップ・サーバーを追加できます。
  • 3fの場合:
    • 「セキュリティ・プロトコル」で、「SASL over Plaintext」を選択します。
    • ユーザー名とパスワードを入力してください

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レポート・ファイルにこのメッセージが表示された場合は、次の操作を実行できます。

  • ターゲット・トピックが存在することを確認するか、ターゲットKafka設定内で自動トピック作成が有効になっていることを確認してください。
  • トラフィックをブロックするファイアウォール・ルールがないことを確認します。
  • プライベート・エンドポイントを使用してOCIでKafkaを実行している場合は、server.propertiesおよびKafka接続で内部FQDNをブートストラップ・サーバーとして使用してください。
  • プライベート・エンドポイントを使用してConfluent Cloudに接続する場合:
    • DNSゾーンおよびDNSレコードがOCIとターゲットのサード・パーティ・クラウドの両方で正しく構成されていることを確認します。
    • OCIとターゲット・クラウド間のネットワーク接続が正常に機能することを確認します。
    • サード・パーティ・クラウドに接続された同じサブネットで実行されているOCI VMから、OpenSSL (openssl s_client -connect <bootstrap>)を使用してターゲットConfluent Cloudに接続できることをテストします。
    • サード・パーティ・クラウドに接続されているのと同じサブネット内で、OCIで実行されているKafkaクライアントからメッセージをパブリッシュまたは消費できることをテストします。失敗した場合は、OCIとサード・パーティ・クラウドの両方でネットワーク設定を確認します。