REST プロデューサー API の使用

Event Streams が提供する REST API は、既存のシステムを Event Streams Kafka クラスターに接続する際に役立ちます。 API を使用すると、RESTful API をサポートする任意のシステムに Event Streams を統合できます。

REST プロデューサー API は、 Event Streams 標準プランおよびエンタープライズ・プランの一部としてのみ使用可能です。

REST プロデューサー API はスケーラブルな REST インターフェースであり、セキュアな HTTP エンドポイントを経由して Event Streams にメッセージをプロデュースするために使用します。 イベント・データを Event Streamsに送信し、 Kafka テクノロジーを使用してデータ・フィードを処理し、 Event Streams 機能を利用してデータを管理します。

API を使用して既存のシステムを Event Streams に接続します。 システムから Event Streams へのプロデュース要求を作成します。これには、メッセージ・キー、ヘッダー、およびメッセージの書き込み先のトピックの指定などが含まれます。

REST Producer API へのアクセス

サービス・インスタンスのサービス資格情報オブジェクトまたはサービス・キーから、 API に接続するために必要な URL と資格情報の詳細を取得する必要があります。 これらのオブジェクトの作成について詳しくは、Event Streams への接続を参照してください。

API のエンドポイントの URL は、kafka_http_urlプロパティーで指定されます。

認証

サポートされる認証メカニズムは、ベアラー・トークンの使用です。 IBM Cloud CLI を使用してトークンを取得するには、まず IBM Cloud にログインしてから、以下のコマンドを実行します。

ibmcloud iam oauth-tokens

このトークンを、Bearer<token>の形式で HTTP 要求の許可ヘッダーに入れます。 API キーと JWT トークンの両方がサポートされています。

REST プロデューサー API を使用したメッセージの生成

Producer API の v2 エンドポイントを使用して、text、 binary、 JSON、avro タイプのメッセージをトピックに送信します。 v2 エンドポイントでは、avro データ・タイプのスキーマを指定することにより、 Event Streams スキーマ・レジストリーを使用できます。

以下のコードは、curl を使用して text タイプのメッセージを送信する例を示しています。

curl -v -X POST \
-H "Authorization: Bearer $token" -H "Content-Type: application/json" -H "Accept: application/json" \
-d '{
  "headers": [
    {
      "name": "colour",
      "value": "YmxhY2s="
    }
  ],
  "key": {
    "type": "text",
    "data": "Test Key"
  },
  "value": {
    "type": "text",
    "data": "Test Value"
  }
}' \
"$kafka_http_url/v2/topics/$topic_name/records"

API について詳しくは、 Event Streams REST プロデューサー API リファレンスを参照してください。

スキーマに準拠するメッセージの生成

REST プロデューサー API の v2 エンドポイントを使用すると、メッセージ・キーと値がスキーマに準拠するようにメッセージを生成できます。 キーと値に異なるスキーマを指定できます。 サポートされるシリアライザーは confluent で、サポートされるデータ・タイプは avro です。 スキーマが作成され、Event Streams スキーマ・レジストリーに保管されます。 詳しくは、 Event Streams スキーマ・レジストリー を参照してください。

許容されるスキーマ命名方針は以下のとおりです。

  • トピック命名方針: スキーマ成果物 ID を生成する際に、トピックの名前が使用されます。 ID の形式は、キーの場合は「<topicName>-key」、値の場合は「<topicName>-value」です。ここで、topicName はトピックの名前です。

  • レコード命名方針: スキーマ成果物 ID を生成する際に、スキーマ内のレコードの名前が使用されます。 ID の形式は、キーの場合は「<composite-recordName>-key」、値の場合は「<composite-recordName>-value」です。 スキーマ名前空間フィールドが指定されている場合、composite-recordName は、"¥ < namespace¥>. ¥ <recordName¥>" の値を取ります。それ以外の場合は、"¥ <recordName¥>" の値を取ります。

  • TopicRecord 命名方針: トピックの名前とレコードの名前の両方が、スキーマ成果物 ID の派生に使用されます。 ID の形式は、キーの場合は「<topicName>-<recordName>-key」、値の場合は「"<topicName>-<composite-recordName>-value」です。ここで、topicName はトピックの名前です。 スキーマ名前空間フィールドが指定されている場合、composite-recordName は、"¥ < namespace¥>. ¥ <recordName¥>" の値を取ります。それ以外の場合は、"¥ <recordName¥>" の値を取ります。

以下のコードは、curl を使用して schema フィールドで指定されたスキーマに準拠するメッセージを送信する例を示しています。

curl -v -X POST \
-H "Authorization: Bearer $token" -H "Content-Type: application/json" -H "Accept: application/json" \
-d '{
  "value": {
    "type": "avro",
    "schema": "{\"namespace\": \"com.eventstreams.samples\",\"type\": \"record\",\"name\": \"recordValueName\",\"fields\": [{\"name\": \"valueName\", \"type\": \"string\"}]}",
    "schema_name_strategy": "record",
    "data": "{\"valueName\": \"sampleValueName\"}"
  }
}' \
"$kafka_http_url/v2/topics/$topic_name/records?serializer=confluent"

既存のエンドポイントから REST Producer API の v2 エンドポイントへのマイグレーション

API 標準との整合性を向上させるために、 v2 エンドポイントに対していくつかの改善が行われました。 これらの改善を最大限に活用するには、既存のアプリケーションに変更を加えます。

以下の考慮事項は、マイグレーションの計画に役立ちます。

  1. REST Producer API へのアクセス:

    v2 エンドポイントには、既存の URL と同じ方法でアクセスできます。つまり、サービス・インスタンスの kafka_http_url プロパティーの値を取得します。 使用するパスは /v2/topics/<topic_name>/records です。

    Example URL: https://service-instance-adsf1234asdf1234asdf1234-0000.us-south.containers.appdomain.cloud/v2/topics/topic_name/records

  2. 認証:

    サポートされる認証メカニズムはベアラー・トークンです。 セキュリティーを強化するために、API キーを使用した基本認証は受け入れられなくなりました。

    Example Header:  -H "Authorization: Bearer $token"

  3. ヘッダー:

    Content-Type ヘッダーと Accept ヘッダーを application/json に設定します。

    Example Headers:  -H "Content-Type: application/json" -H "Accept: application/json"

  4. ペイロード:

    v2 エンドポイントのペイロードを JSON 形式で指定します。 ペイロードには、メッセージ・キー、ヘッダー、およびデータを定義できます。 base64 でエンコードされた値を使用して、リストの形式でヘッダーを指定できます。 ただし、メッセージ・キーとヘッダーはオプションです。

    Example payload:
{
  "headers": [
  {
   "name": "colour",
   "value": "YmxhY2s="
  },
  "key": {
   "type": "text",
   "data": "Test Key"
  },
  "value": {
   "type": "text",
   "data": "Test Value"
  }
 }

    キーおよび値オブジェクトの下で、フィールド・タイプに対してサポートされる以下のいずれかのデータ・タイプを指定する必要があります。

    a. テキスト: 提供されるデータは、文字の線形シーケンスで構成されるプレーン・テキスト形式として検証されます。

    b. バイナリー: 提供されるデータは、 base64-encoded バイナリー形式として検証されます。

    c. JSON: 提供されるデータは、 JSON 形式として検証されます。

    d. Avro: データは Apache Avro データ形式として検証されます。

  5. エラー応答:

    エラー応答には、 trace、 error message、 error code、 more_info、および target の各プロパティーが含まれます。

    Example error response:
{
 "trace": "a222e93c-e5f9-435d-b275-5d4919ea87ed",
 "error": {
  "code": "invalid_type",
  "message": "'type' field in 'value' object is required ...",
  "more_info": "https://cloud.ibm.com/apidocs/event-streams/restproducer_v2",
  "target": {
   "type": "field",
   "name": "type"
  }
 }
}

制限

REST プロデューサー API を使用する場合、要求ペイロードとして渡されるメッセージの最大サイズに制限があります。 ペイロードの最大サイズは 64 K に制限されています。

API リファレンス

v2 エンドポイント API の詳細については、 Event Streams REST Producer v2 API リファレンスを参照してください。

既存のエンドポイント API の詳細については、 Event Streams REST プロデューサー API リファレンスを参照してください。