Apache Kafka コンシューマーコネクタの構成

  • リリースバージョン: Washingtondc
  • 更新日 2024年02月01日
  • 読む7読むのに数分

    • Apache KafkaコネクターによってApache Kafka収集されたストリーミングメッセージからイベントを作成するように、コンシューマーコネクタインスタンスを設定します。

    始める前に

    イベント管理 Connectors (sn_em_connector) プラグインが Now Platform インスタンスにインストールされていることを確認します。

    必要なロール：evt_mgmt_admin

    手順

    1. 移動先 すべて > イベント管理 > 統合 > コネクタインスタンス.
    2. [新規] を選択します。
    3. フォームのフィールドに入力します。
      表 : 1. [コネクタインスタンス] フォーム
      フィールド 値
      名前 Kafka コンシューマーコネクタのわかりやすい一意の名前。
      説明 Kafka コンシューマーイベント収集インスタンスの使用についての説明です。
      コネクタ定義 必要なコネクタ定義の名前。この場合は Kafka Consumer です。
      ホスト IP ホスト IP。
      注:
      作成プロセスを完了するにはこのフィールドに値を含める必要があるため、一時値としてプレースホルダー 1.1.1.1 を入力する必要があります。
      前回のイベント収集実行時間 このフィールドは自動的に最後のランタイム値に設定されます。
      前回のイベント収集ステータス このフィールドは自動的に最後のランタイムステータスに設定されます。
      イベント収集スケジュール (秒) システムが Kafka コンシューマーからの新しいイベントをチェックする頻度 (秒単位)。デフォルト値は 60 秒です。
      前回エラーメッセージ このフィールドは自動的に最後のエラーメッセージに設定されます。
    4. PLAINTEXT または SASL_PLAINTEXT の場合は、[資格情報] フィールドで基本認証 資格情報 を作成します。
      SASL_PLAINTEXT認証で新しい Kafka プルコネクタインスタンスには、Kafka SSL 資格情報タイプを使用します。手順 5 に進みます。

      Kafka サーバーの sasl.enabled.mechanism または sasl.mechanism パラメーターの値が PLAIN と等しくない場合は、Kafka SSL 資格情報タイプを使用します。手順 5 に進みます。

      1. [資格情報] フィールドの横にある検索アイコン 検索アイコン を選択します。
      2. [資格情報] 画面で、[新規] を選択します。
      3. 使用可能な資格情報のリストで、[基本認証資格情報] を選択します。
      4. [基本認証] 画面で、資格情報を必要としない PLAINTEXT 認証か、資格情報を必要とする SASL_PLAINTEXT 認証のいずれかの情報を入力します。
        • PLAINTEXT 認証の場合は、[名前] フィールドに任意の値を入力し、[ユーザー名] フィールドと [パスワード] フィールドを空のままにします。
        • SASL_PLAINTEXT 認証の場合は、[名前] フィールドに資格情報の名前を指定します (先頭に SASL を付けることをお勧めします)。[ユーザー名] フィールドと [パスワード] フィールドにユーザー名とパスワードを入力します。
      5. [送信] を選択します。
    5. SASL_PLAINTEXT、SSL、または SASL_SSL 認証資格情報の場合は、[ 資格情報 ] フィールドで Kafka SSL 資格情報を作成します。
      1. [資格情報] フィールドの横にある検索アイコン 検索アイコン を選択します。
      2. [資格情報] 画面で、[新規] を選択します。
      3. 使用可能な資格情報のリストから、[ Kafka SSL 資格情報] を選択します。
      4. [Kafka SSL 認証] 画面で、[Kafka SSL 資格情報] フィールドに入力します。
        Kafka SSL 資格情報フィールドの説明については、「 」を参照してください。
        注:
        [ ホスト名の検証を無効にする] フィールドをオフにした場合は、[追加の Kafka コンシューマープロパティ] フィールドに、 Kafka サーバーパラメーター値と同じパラメーターの値 ssl.endpoint.identification.algorithm を入力します。例： 
        ssl.enabled.protocols=TLSv1.2,TLSv1.1,TLSv1;
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username=<username> password=<password>;
sasl.enabled.mechanisms=PLAIN;ssl.endpoint.identification.algorithm=;
      5. [送信] を選択します。
    6. フォームヘッダーを右クリックし、[保存] を選択します。
    7. コネクタインスタンス値を示すテーブルで、Kafka セットアップと Kafka トピックから受信したメッセージ (JSON ペイロード) に基づいて、設定されたコネクタインスタンス値を確認します。

      必要に応じて、コネクタインスタンスの値を変更します。

      表 : 2. コネクタインスタンスの値テーブル
      フィールド 説明
      authentication_type

      Kafka コンシューマーの認証タイプ。現在サポートされている値は次のとおりです。

      • PLAINTEXT (認証なし) (デフォルト)

      • SASL_PLAINTEXT (基本認証 – ユーザー名とパスワード)
      bootstrap_servers

      Kafka クラスターとの接続を確立するサーバー。このフィールドの値は、host1:port1、host2:port2 などの形式にする必要があります (カンマ区切り)。

      このフィールドは必須です。
      consumer_group_name

      コンシューマーグループ名

      同じ Kafka トピックが 2 つの異なるインスタンスから使用されている場合は、必ず異なるコンシューマーグループ名を使用して、すべてのイベントが両方のインスタンスでキャプチャされるようにしてください。

      このフィールドは必須です。
      time_of_event_field

      イベントの時間を含む JSON ペイロード/メッセージのフィールドの名前。

      JSON ペイロード/メッセージのフィールドにイベントの時間が含まれている場合は、その値がこのパラメーターの値になります。

      時間フィールドの値が指定されていない場合、ServiceNow インスタンスでイベントを受信した時間が time_of_event として設定されます。

      time_of_event フィールドは、JSON ペイロード/メッセージの第 1 レベルのフィールドである必要があります。ネストされたフィールド名は使用できません。
      timezone

      JSON ペイロード/メッセージの time_of_event フィールドのタイムゾーン。

      time_of_event フィールドが空の場合、timezone の値は使用されません。

      例えば、ペイロード/メッセージの time_of_event が IST タイムゾーンからのものである場合、その値は GMT + 0530 になります。

      デフォルト：GMT
      date_format

      JSON ペイロード/メッセージの time_of_event フィールドの date_time 形式。

      時間フィールドの値が指定されていない場合、インスタンスでイベントを受信した時間が GMT 時刻として使用されます。time_of_event フィールドが空の場合、date_time の値は使用されません。

      デフォルト：yyyy-MM-dd HH:mm:sss
      debug デバッグログを表示するオプション。デフォルト値：false
      logPayloadForDebug JSON ペイロード関連のログを表示するオプション。デフォルト値：false
      topic

      メッセージのフェッチ元の Kafka トピック。

      このフィールドは必須です。
    8. [コネクタの MID Server (MID Servers for Connectors)] セクションで、稼働している有効な MID サーバー を指定します。
      注:
      複数の MID サーバー を設定できます。最初のサーバーが停止すると、次の MID サーバー が使用されます。その MID サーバー が利用できない場合は、さらに次のサーバーが選択され、以下同様に続きます。MID サーバー は、その詳細情報が [コネクタの MID Server] セクションに入力された順にソートされます。

      を指定しない場合は MID サーバー、IP 範囲が一致する使用可能な MID サーバー ものが使用されます。

    9. フォームヘッダーを右クリックし、[保存] を選択します。
    10. MID サーバー と Kafka コンシューマーコネクタの間の接続をテストします。
      1. [コネクタのテスト] を選択します。
      2. テスト接続に失敗した場合は、資格情報が有効かどうか、およびネットワークが MID サーバー から Kafka ブローカーに接続されていることを確認します。
        注:
        Kafka トピック名の検証は 、[コネクタのテスト] の検証でのみ行われます。
    11. テストに成功したら、[アクティブ] チェックボックスをオンにしてコネクターインスタンスをアクティブにします。
    12. [Update] ボタンを選択します。

    次のタスク

    コネクタの作成後、フィールドをマッピングする必要があります。詳細については、「Kafka メッセージペイロード属性からアラートフィールドへのマッピング」を参照してください。