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

  • リリースバージョン: Zurich
  • 更新日 2025年07月31日
  • 所要時間：7分

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

    始める前に

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

    必要なロール：evt_mgmt_admin

    手順

    1. 移動先 すべて > イベント管理 > 統合 > コネクタインスタンス.
    2. [New (新規)] を選択します。
    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.mechanisms または 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 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。デバッグメッセージを表示するには、true を指定します。
      logPayloadForDebug ペイロード関連のデバッグメッセージを表示します。デフォルト値：false。ペイロード関連デバッグメッセージを表示するには、true を指定します。
      トピック

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

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

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

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

    次のタスク

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