Apache Kafkaへの接続

OCI GoldenGateソースまたはターゲットとして使用するApache Kafkaへの接続の作成について学習します。

開始する前に

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

• OCI GoldenGateがソースおよびターゲットに接続する方法を確認します。

• シークレットの使用、ボールトの使用、シークレット・バンドルの読取りなど、セキュアなVaultおよびシークレット・アクセスを有効にするために必要なポリシーを構成します。詳細は、最小推奨ポリシーを参照してください。

• TLS/mTLSを使用している場合は、接続で使用するJKSトラストストアまたはトラストストアおよびキーストアをPKCS12形式に変換します。

  1. JDKのkeytoolユーティリティを使用して、PKCS12形式に変換します。

     キーストアの場合、次の例に示すように、keytoolユーティリティによってパスワードの入力が求められます。

     keytool -importkeystore -srckeystore [MY_KEYSTORE.jks] -destkeystore [MY_KEYSTORE.p12] -srcstoretype JKS -deststoretype PKCS12 -deststorepass [PASSWORD_PKCS12]

     トラストストアの場合、次の例に示すように、keytoolユーティリティによってパスワードの入力が求められます。

     keytool -importkeystore -srckeystore [MY_TRUSTSTORE.jks] -destkeystore [MY_TRUSTSTORE.p12] -srcstoretype JKS -deststoretype PKCS12 -deststorepass [PASSWORD_PKCS12]

  2. キーストアおよびトラストストアを変換した後、次の行をKafkaプロデューサ・プロパティ・ファイルに追加し、変更を保存します。

     ssl.truststore.type=PKCS12
     ssl.keystore.type=PKCS12

  3. 「拡張オプション」の「設定」セクションで、接続のプロデューサ・プロパティにファイルをアップロードします。

ソース接続の作成

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

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

         ノート:

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

         ヒント :

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

          ノート:

          • 専用接続が7日間未割当てのままの場合、サービスはそれを共有接続に変換します。
          • Oracle GoldenGate接続についてさらに学習します。
      • 設定
        • この接続をExtractに使用する場合は、コンシューマまたはプロデューサ・プロパティ・ファイル(.properties)をドラッグ・アンド・ドロップするか選択します。
        • Kafkaから取得するには、次のいずれかのデシリアライザまたはコンバータを使用してKafkaコンシューマ・プロパティ・ファイル(.properties)を作成します。
          • 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. 「作成」をクリックします。

接続が作成されると、「接続」リストに表示されます。接続をデプロイメントに割り当てて、レプリケーションのソースまたはターゲットとして使用するようにしてください。

ターゲット接続の作成

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

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

         ノート:

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

         ヒント :

         クラスタ内のすべてのノードに、プライベート・エンドポイントでのトラバースを許可するFQDNが必要です。
      2. (オプション)別のブートストラップ・サーバーを追加するには、「+ブートストラップ・サーバー」をクリックします。
   7. 「セキュリティ・プロトコル」で、次のいずれかを選択し、対応するフィールドに入力します:
      • 平文(Plaintext)
      • SASL over plaintext
      • 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. 「作成」をクリックします。

接続が作成されると、「接続」リストに表示されます。接続をデプロイメントに割り当てて、レプリケーションのソースまたはターゲットとして使用するようにしてください。

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とサード・パーティ・クラウドの両方でネットワーク設定を確認します。

---

タイトルおよび著作権情報

Copyright ©2022,

Oracle and/or its affiliates.