Event Streams スキーマ・レジストリーの使用

スキーマ・レジストリーは、スキーマを管理および検証するための一元化されたリポジトリーを提供します。スキーマ・レジストリー内のスキーマは、イベントを生成するプログラムが、それらのイベントをコンシュームする他のプログラムに提供する明示的な契約を提供します。

スキーマの概要

Apache Kafka はすべてのデータを処理できますが、メッセージ内の情報は検証しません。ただし、データを効率的に処理するには、通常、特定の情報が特定のフォーマットで含まれている必要があります。スキーマを使用して、メッセージ内のデータの構造を定義し、プロデューサーとコンシューマーの両方で適切な構造が使用されるようにすることができます。

スキーマは、事前定義された構造に準拠するデータをプロデューサーで作成するのに役立ち、存在する必要があるフィールドとそのタイプを定義します。この定義は、コンシューマーがそのデータを解析して正しく解釈するのに役立ちます。Event Streams エンタープライズ・プランは、スキーマをサポートし、スキーマを使用および管理するためのスキーマ・レジストリーを含んでいます。

一般的に、特定トピックのメッセージはすべて同じスキーマを使用します。メッセージのキーと値は、それぞれをスキーマで記述できます。

スキーマ・レジストリー

スキーマは、Event Streams スキーマ・レジストリー内に保管されます。スキーマ・レジストリーは、バージョン付きのスキーマ履歴を保管することに加えて、スキーマを取得するためのインターフェースも提供します。各エンタープライズ・プラン Event Streams インスタンスには、独自のスキーマ・レジストリーがあります。 1 つのエンタープライズ・インスタンスには、最大 1000 個のスキーマを保管できます。

生産者と消費者は、Schema Registryに保存されている指定のスキーマに対してデータを検証します（ Kafka の仲介業者を通すことに加えて）。このように、メッセージでスキーマを転送する必要がないため、メッセージをより小さくすることができます。

Apache Avro データ・フォーマット

スキーマは、 Apache Kafkaで一般的に使用されるオープン・ソースのデータ・シリアライゼーション・テクノロジーである Apache Avro を使用して定義されます。コンパクトなバイナリー・フォーマット、または冗長ではありながら人間が理解できる JSON フォーマットのいずれかを使用して、効率的なデータ・エンコード・フォーマットを提供します。

Event Streams スキーマ・レジストリーは、Apache Avro データ・フォーマットを使用します。 Avro フォーマットで送信されるメッセージには、データおよび使用されるスキーマの固有 ID が含まれます。 ID は、メッセージに使用するレジストリー内のスキーマを指定します。

Avro は、プリミティブ型 (null、boolean、int、long、float、double、bytes、および string) と複合型 (record、enum、array、map、union、および fixed) などの幅広いデータ型をサポートします。

で送信されたメッセージの表現を示す図

シリアライゼーションおよびデシリアライゼーション

生成アプリケーションは、シリアライザを使用して特定のスキーマに準拠するメッセージを生成します。前述のように、メッセージには、スキーマ ID と共に Avro フォーマットのデータが含まれています。

次に、消費アプリケーションは、同じスキーマを使用してシリアル化されたメッセージを消費するために、デシリアライザを使用します。消費者が Avro フォーマットで送信されたメッセージを読み取ると、デシリアライザーがメッセージ内のスキーマの識別子を見つけ、データをデシリアライズするためにスキーマレジストリからスキーマを取得します。

このプロセスにより、メッセージ内のデータが要求された構造に適合していることを効率的に確認することができます。

Event Streams スキーマレジストリは、 Kafka AVRO シリアライザーおよびデシリアライザーをサポートしています。

シリアル化とデシリアル化の図。 — シリアライザーとデシリアライザーに適合する場所を示す図

バージョンおよび互換性

スキーマを追加するたびに、また同じスキーマのすべての後続バージョンについても、 Event Streams はフォーマットを自動的に検証し、問題があればスキーマを拒否することができます。時間の経過に伴って変化する要件に応じてスキーマを発展させることができます。既存のスキーマの新しいバージョンを作成すると、スキーマレジストリが新しいバージョンが既存のバージョンと互換性があることを保証します。つまり、既存のバージョンを使用するプロデューサーとコンシューマーは、新しいバージョンによって影響を受けないということです。

スキーマは、スキーマのセマンティクスに影響しない方法でのみスキーマが異なる場合に、スキーマの重複を避けるために比較されます。場合によっては、スキーマ内の JSON プロパティーの順序付けが、データのエンコードおよびデコードにスキーマがどのように使用されるかにとって重要になることがありますが、それ以外の場合は関係ない可能性があります。

例えば、レコード・スキーマの name プロパティーは、エンコードおよびデコード・プロセスの一部として使用されないため、レコード JSON オブジェクト内の任意の場所に配置できます。これらのバリエーションはすべて同じスキーマと見なされます。

レコード・スキーマの JSON 内の fields プロパティーは、その順序が重要な場合です。 Avro 仕様では、エンコードおよびデコード操作に使用されるスキーマに表示される順序で、レコードのフィールドがエンコードおよびデコードされる必要があります。

例として、以下の3つのスキーマを考えてみましょう。

スキーマ 1

{
  "type": "record",
  "name": "book",
  "fields": [
    {
      "name": "title",
      "type": "string"
    },
    {
      "name": "author",
      "type": "string"
    }
  ]
}

スキーマ 2

{
  "type": "record",
  "name": "book",
  "fields": [
    {
      "name": "author",
      "type": "string"
    },
    {
      "name": "title",
      "type": "string"
    }
  ]
}

スキーマ 3

{
  "type": "record",
  "name": "book",
  "fields": [
    {
      "type": "string"
      "name": "author",
    },
    {
      "type": "string"
      "name": "title",
    }
  ]
}

スキーマ 1 とスキーマ 2 は別個のスキーマであり、レジストリーはそれらを別個のスキーマとして保管します。 author フィールドと title フィールドは異なる順序でリストされるため、相互に交換して使用することはできません。デコード・プロセスでスキーマ 2 が使用された場合、スキーマ 1 でエンコードされたデータは正しくデコードされません。

SerDes を使用して、スキーマ 1、スキーマ 2、およびスキーマ 3 の順序で新規スキーマを作成すると、結果は 2 つの新規スキーマになります。スキーマ 1 とスキーマ 2 は異なりますが、スキーマ 3 はスキーマ 2 と同等です。

REST API を使用してスキーマを作成する場合、スキーマが一致すると見なされるのは、すべての属性の順序付けフィールドと記述フィールドを含め、それらのスキーマがテキスト的に同じである場合のみです。これは、スキーマ 3 を別のスキーマにしたい場合に使用できるようにするためです。

スキーマ・レジストリーの有効化

Event Streams エンタープライズ・プラン・サービス・インスタンスの場合、スキーマ・レジストリーはデフォルトで有効になっています。スキーマ・レジストリーは、他の Event Streams プランでは使用できません。

スキーマ・レジストリーへのアクセス

スキーマレジストリにアクセスするには、お客様のサービスのサービス認証情報に記載されているスキーマレジストリの URL が必要です。これらの認証情報を UI で表示するには、サービスインスタンスをクリックし、左側のナビゲーションペインで 「サービス認証情報」 を選択し、表にリストされたサービス認証情報のいずれかの隣にある 「認証情報の表示」リンクをクリックします

にアクセスするために必要な資格情報フィールドの表現を示す図

kafka_http_url の値は、スキーマレジストリの URL でもあります。

認証

スキーマ・レジストリーにアクセスするには、レジストリーでの認証に使用できる資格情報のセットも必要です。 API キーを使用した基本認証とベアラー・トークン認証の 2 つのオプションがあります。

本書の例では、API キーの使用を示していますが、どちらのオプションも使用できます。

API キーによる認証

サービス資格情報には、スキーマ・レジストリーでの認証のための資格情報として使用できる apikey があります。

サービスIDから付与されたAPIキーを使用して認証することもできます。ただし、サービスIDが少なくとも「リーダー」ロールのアクセスを Event Streams インスタンスに許可するポリシーを持っている場合に限ります。これは、より柔軟性のあるアプローチであり、他の複数のユーザーまたはチームにアクセス権限を付与する場合にはより適切な選択肢です。詳しくは、ヘルプ・トピック Event Streams リソースへのアクセスの管理を参照してください。

APIキーは、 HTTP の基本認証ヘッダーのパスワード部分として提供されます。ヘッダーのユーザー名部分は、「token」という語です。

使用する curl コマンドは以下のとおりです。ここで、$APIKEY は API キーで置き換えられます。

curl -u token:$APIKEY ...

ベアラー・トークンによる認証

また、システムIDやユーザーの認証情報として、無記名トークンを使用することも可能です。これは、API キーを公開する可能性が低く、しばらくするとベアラー・トークンが自動的に期限切れになるため、一般的にはより安全な方法です。

トークンを取得するには、 IBM Cloud CLI ibmcloud iam oauth-tokens コマンドを使用してトークンを生成します。このトークンを「Authorization: Bearer $TOKEN」という形式で HTTP ヘッダーに含めます。$TOKENはベアートークンです

curl -H "Authorization: Bearer $TOKEN" ...

他のスキーマ・レジストリーからのデータのインポート

他のスキーマ・レジストリーからエクスポートされたスキーマ・レジストリーにデータをインポートできます。データがインポートされると、各成果物バージョンに関連付けられているグローバル ID が保持されます。つまり、同じスキーマ・グローバル ID 値を使用して、 Kafka に既に保管されているデータを引き続き使用できます。

Event Streams CLI は、以下の例のように、Apicurio レジストリーのインポートおよびエクスポート形式を使用したデータのインポートをサポートします。

ibmcloud es schema-import import.zip

Apicurio レジストリー exportConfluent ユーティリティーを使用して、インポートするデータを生成できます。このユーティリティーは、Confluent スキーマ・レジストリーからデータをエクスポートします。 Event Streamsバージョンでテスト済み2.6.xこのユーティリティの。

インポート中の成果物バージョンと同じグローバル ID を持つエントリーが Event Streams スキーマ・レジストリーに既に存在する場合、インポート操作は失敗し、続行する場合は成果物バージョンの削除を求めるプロンプトが出されます。

スキーマ・レジストリー REST エンドポイント

REST API は、以下の 4 つの主要機能を提供します。

スキーマの作成、読み取り、および削除
スキーマの個々のバージョンの作成、読み取り、および削除
レジストリーのグローバル互換性ルールの読み取りおよび更新
個々のスキーマに適用される互換性ルールの作成、読み取り、更新、および削除

成果物、成果物バージョン、およびルールの作成、更新、削除など、スキーマ・バージョンを変更するアクションの場合は、Activity Tracker イベントが生成され、アクションが報告されます。詳しくは、Activity Tracker イベントを参照してください。

エラー

エラー状態が発生した場合、スキーマレジストリは non-2XX の範囲 HTTP のステータスコードを返します。レスポンスの本文には、以下の形式のJSONオブジェクトが含まれます

{
    "error_code":404,
    "message":"No artifact with id 'my-schema' might be found."
}

エラー JSON オブジェクトのプロパティーは以下のとおりです。

プロパティー名	説明
error_code	応答の HTTP 状況コード。
メッセージ	問題の原因の説明。
インシデント	このフィールドは、このエラーがスキーマ・レジストリーでの問題の結果である場合にのみ含まれます。この値は、IBM サービスで、レジストリーによってキャプチャーされた診断情報に要求を関連付けるために使用できます。

スキーマ状態の設定

このエンドポイントは、レジストリー内のスキーマの状態を ENABLED または DISABLED に設定するために使用されます。スキーマの状態は、 /artifacts/{schema-id}/state エンドポイントに PUT 要求を発行することによって設定できます ( {schema-id} はスキーマの ID です)。リクエストが成功した場合、空の応答とステータスコード204（コンテンツなし）が返されます。

curl 要求の例:

curl -u token:$APIKEY –X PUT $URL/artifacts/my-schema/state -d '{"state": "DISABLED"}'

スキーマ状態を設定するには、以下が必要です。

変更されたスキーマに一致するスキーマリソースへのマネージャーロールのアクセス。

スキーマ・バージョン状態の設定

このエンドポイントは、レジストリー内のスキーマ・バージョンの状態を ENABLED または DISABLED に設定するために使用されます。スキーマ・バージョンの状態は、 /artifacts/{schema-id}/versions/{version}/state エンドポイントに PUT 要求を発行することによって設定できます ( {schema-id} はスキーマの ID、 {version} はスキーマ・バージョンのバージョン番号です)。リクエストが成功した場合、空の応答とステータスコード204（コンテンツなし）が返されます。

curl 要求の例:

curl -u token:$APIKEY –X PUT $URL/artifacts/my-schema/versions/1/state -d '{"state": "DISABLED"}'

スキーマ・バージョンの状態を設定するには、以下が必要です。

マネージャーロールは、変更対象のスキーマに一致するスキーマリソースへのアクセス権限を持つ。

スキーマの作成

このエンドポイントは、スキーマをレジストリーに保管するために使用されます。スキーマ・データは、POST 要求の本文として送信されます。 ‘X-Registry-ArtifactId' リクエストヘッダーを使用することで、スキーマのIDを含めることができます。リクエストにこのヘッダーが含まれていない場合、IDが生成されます。コンテンツ・タイプ・ヘッダーは、「application/json」に設定する必要があります。

curl 要求の例:

curl -u token:$APIKEY -H 'Content-Type: application/json' -H 'X-Registry-ArtifactId: my-schema' $URL/artifacts -d '{"type":"record","name":"Citizen","fields":[{"name": "firstName","type":"string"},{"name":"lastName","type":"string"},{"name":"age","type":"int"},{"name":"phoneNumber","type":"string"}]}'

応答の例:

{"id":"my-schema","type":"AVRO","version":1,"createdBy":"","createdOn":1579267788258,"modifiedBy":"","modifiedOn":1579267788258,"globalId":75}

スキーマを作成するには、少なくとも以下の両方が必要です。

Event Streams クラスター・リソース・タイプに対するリーダー役割アクセス
作成されたスキーマに一致するスキーマリソースへのライターロールのアクセス。

Activity Tracker イベントが生成され、アクションが報告されます。詳しくは、Activity Tracker イベントを参照してください。

スキーマのリスト

/artifacts エンドポイントに GET リクエストを送信することで、レジストリに保存されているすべてのスキーマの ID のリストを生成できます。 jsonformat パラメーターを使用して応答をフォーマット設定できます ( string および object フォーマットのみがサポートされます)。ストリング・フォーマットはデフォルトであり、成果物 ID (ストリング) の配列を返します。このオプションが設定されている場合、有効な成果物のみが配列に組み込まれます。オブジェクト・フォーマットは、配列を含む JSON オブジェクトを返します。配列内の各項目は、レジストリー内の成果物に対応しています。このオプションが設定されている場合、有効な成果物と無効な成果物の両方が返されます。

curl 要求の例:

curl -u token:$APIKEY $URL/artifacts

または

curl -u token:$APIKEY $URL/artifacts?jsonformat=string

または

curl -u token:$APIKEY $URL/artifacts?jsonformat=object

jsonformat が string であるか、指定されていない場合の応答例 (デフォルトは string):

["my-schema-2","my-schema-4"]

jsonformat がオブジェクトの場合の応答例:

{"artifacts":[{"id":"my-schema","state":"DISABLED"},{"id":"my-schema-2","state":"ENABLED"},{"id":"my-schema-3","state":"DISABLED"},{"id":"my-schema-4","state":"ENABLED"}],"count":4}

スキーマをリストするには、少なくとも以下が必要です。

Event Streams クラスター・リソース・タイプに対するリーダー役割アクセス

スキーマの状態と削除

スキーマの削除は、2 段階のプロセスです。削除の最初の段階では、スキーマはレジストリーに保持されますが、一部の操作では非表示になります。 2 番目のステージはスキーマを完全に削除しますが、適用できるのは最初のステージの後に限られます。 2 段階の削除プロセスは、成果物レベルで適用され、バージョン・レベルでも適用されます。

削除の 2 つのステージは、成果物とバージョンの両方に関連付けられた使用可能または使用不可の状況を持つこと (最初のステージ)、およびリソースとバージョンの API を削除すること (2 番目のステージ) によって行われます。

無効にされた成果物またはバージョンは、成果物またはバージョンをリストする操作によって返される「状態」プロパティーを使用するか、成果物またはバージョンの詳細を取得することによって検出できます。無効にされたスキーマは、エンタープライズ・インスタンスごとに 1000 スキーマのスキーマ割り当て量にカウントされます。

スキーマの削除

スキーマは、 /artifacts/{schema-id} エンドポイントに DELETE リクエストを発行することでレジストリから削除されます（ {schema-id} はスキーマのID）。成功すると、空の応答と状況コード 204 (コンテンツなし) が返されます。

curl 要求の例:

curl -u token:$APIKEY -X DELETE $URL/artifacts/my-schema

スキーマを削除するには、少なくとも以下の両方が必要です。

Event Streams クラスター・リソース・タイプに対するリーダー役割アクセス
削除されたスキーマに一致するスキーマリソースへのマネージャーロールのアクセス。

Activity Tracker イベントが生成され、アクションが報告されます。詳しくは、Activity Tracker イベントを参照してください。

スキーマの新規バージョンの作成

スキーマの新しいバージョンを作成するには、 /artifacts/{schema-id}/versions エンドポイントに POST リクエストを送信します（ {schema-id} はスキーマのID）。要求の本文には、スキーマの新規バージョンが含まれている必要があります。

リクエストが成功した場合、新しいスキーマは適切なバージョン番号とともにスキーマの最新バージョンとして作成され、ステータスコード200（OK）の応答が返され、新しいバージョン（バージョン番号を含む）を説明するメタデータを含むペイロードが返されます。

curl 要求の例:

curl -u token:$APIKEY -H 'Content-Type: application/json' $URL/artifacts/my-schema/versions -d '{"type":"record","name":"Citizen","fields":[{"name": "firstName","type":"string"},{"name":"lastName","type":"string"},{"name":"age","type":"int"},{"name":"phoneNumber","type":"string"}]}'

応答の例:

{"id":"my-schema","type":"AVRO","version":2,"createdBy":"","createdOn": 1579267978382,"modifiedBy":"","modifiedOn":1579267978382,"globalId":83}

スキーマの新規バージョンを作成するには、少なくとも以下の両方が必要です。

Event Streams クラスター・リソース・タイプに対するリーダー役割アクセス
新しいバージョンを取得するスキーマに一致するスキーマリソースへのライターロールのアクセス。

Activity Tracker イベントが生成され、アクションが報告されます。詳しくは、Activity Tracker イベントを参照してください。

スキーマの最新バージョンの取得

特定のスキーマの最新バージョンを取得するには、 /artifacts/{schema-id} エンドポイントに GET リクエストを送信します（ {schema-id} はスキーマの ID です）。成功すると、スキーマの最新バージョンが応答のペイロードで返されます。

curl 要求の例:

curl -u token:$APIKEY $URL/artifacts/my-schema

応答の例:

{"type":"record","name":"Citizen","fields":[{"name": "firstName","type":"string"},{"name":"lastName","type":"string"},{"name":"age","type":"int"},{"name":"phoneNumber","type":"string"}]}

スキーマの最新バージョンを取得するには、少なくとも以下の両方が必要です。

Event Streams クラスター・リソース・タイプに対するリーダー役割アクセス
取得したスキーマに一致するスキーマリソースへのリーダロールのアクセス。

スキーマの特定バージョンの取得

特定のバージョンのスキーマを取得するには、 /artifacts/{schema-id}/versions/{version} エンドポイントに GET リクエストを送信します（ {schema-id} はスキーマのID、 {version} は取得したい特定のバージョンのバージョン番号です）。成功すると、スキーマの指定したバージョンが応答のペイロードで返されます。

curl 要求の例

curl -u token:$APIKEY $URL/artifacts/my-schema/versions/3

応答の例:

{"type":"record","name":"Citizen","fields":[{"name": "firstName","type":"string"},{"name":"lastName","type":"string"},{"name":"age","type":"int"},{"name":"phoneNumber","type":"string"}]}

スキーマの最新バージョンを取得するには、少なくとも以下の両方が必要です。

Event Streams クラスター・リソース・タイプに対するリーダー役割アクセス
取得したスキーマに一致するスキーマリソースへのリーダロールのアクセス。

スキーマのすべてのバージョンのリスト

現在レジストリに保存されているスキーマのすべてのバージョンをリストアップするには、 /artifacts/{schema-id}/versions エンドポイントに GET リクエストを送信します（ {schema-id} はスキーマの ID です）。成功すると、スキーマのすべての現行のバージョン番号のリストが応答のペイロードで返されます。 jsonformat パラメーターを使用して応答をフォーマット設定できます ( number および object フォーマットのみがサポートされます)。「数値」(デフォルト) を指定した場合、応答は、成果物の有効なバージョンに対応する数値の配列になります (無効なバージョンは省略されます)。これは、現在生成されているエンドポイントと同じ形式です。「object」を指定した場合、応答は、成果物のバージョンを表す JSON オブジェクトの配列を含む JSON オブジェクトになります。有効なバージョンと無効なバージョンの両方がアレイに含まれています。

curl 要求の例:

curl -u token:$APIKEY $URL/artifacts/my-schema/versions

または

curl -u token:$APIKEY $URL/artifacts/my-schema/versions?jsonformat=number

または

curl -u token:$APIKEY $URL/artifacts/my-schema/versions?jsonformat=object

jsonformat が number であるか、指定されていない (デフォルトは number) 場合の応答例:

[1,3,4,6,7]

jsonformat がオブジェクトの場合の応答例:

{"versions":[{"id":1,"state":"ENABLED"},{"id":2,"state":"DISABLED"},{"id":3,"state":"ENABLED"},{"id":4,"state":"ENABLED"},{"id":5,"state":"DISABLED"},{"id":6,"state":"ENABLED"},{"id":7,"state":"ENABLED"}],"count":7}

スキーマの使用可能なバージョンのリストを取得するには、少なくとも以下の両方が必要です。

Event Streams クラスター・リソース・タイプに対するリーダー役割アクセス
取得したスキーマに一致するスキーマリソースへのリーダロールのアクセス。

スキーマのバージョンの削除

スキーマバージョンは、 /artifacts/{schema-id}/versions/{version} エンドポイントにDELETEリクエストを発行することで、レジストリから削除されます（ {schema-id} はスキーマのID、 {version} はスキーマバージョンのバージョン番号です）。成功すると、空の応答と状況コード 204 (コンテンツなし) が返されます。残っている唯一のバージョンのスキーマを削除すると、スキーマも削除されます。

curl 要求の例:

curl -u token:$APIKEY -X DELETE $URL/artifacts/my-schema/versions/3

スキーマ・バージョンを削除するには、少なくとも以下の両方が必要です。

Event Streams クラスター・リソース・タイプに対するリーダー役割アクセス
削除されたスキーマに一致するスキーマリソースへのマネージャーロールのアクセス。

Activity Tracker イベントが生成され、アクションが報告されます。詳しくは、Activity Tracker イベントを参照してください。

スキーマ・バージョンの特定のグローバル固有 ID の取得

特定のグローバル一意IDのスキーマバージョンを取得するには、 /artifacts/{artifactId}/versions/{version}/meta エンドポイントにGETリクエストを送信します（ {artifactId} はアーティファクトのID、 {version} は取得したい特定バージョンのバージョン番号です）。成功した場合、スキーマ・バージョンの特定のグローバル固有 ID が応答のペイロードで返されます。

curl 要求の例:

curl -u token:$APIKEY $URL/artifacts/9030f450-45fb-4750-bb37-771ad49ee0e8/versions/1/meta

応答の例:

{"id":"9030f450-45fb-4750-bb37-771ad49ee0e8","type":"AVRO","version":1,"createdOn":1682340169202,"modifiedOn":1682340169202,"globalId":1}

スキーマ・バージョンのグローバル固有 ID を取得するには、少なくとも以下の両方のタイプのアクセス権限が必要です。

Event Streams クラスター・リソース・タイプに対するリーダー役割アクセス
取得したスキーマに一致するスキーマリソースへのリーダロールのアクセス。

グローバル・ルールの更新

グローバル互換性ルールは、/rules/ {rule-type} エンドポイントに PUT リクエストを発行することで更新できます（ {rule-type} は更新するグローバルルールのタイプを識別します。現在サポートされているタイプは互換性のみです）。リクエストのボディに新しいルール構成を含めます。要求が成功すると、新しく更新されたルール構成が、状況コード 200 (OK) と共に応答のペイロードで返されます。

リクエストボディで送信されるJSONドキュメントは、以下のプロパティを持つ必要があります

プロパティー名	説明
タイプ	常に値 COMPATIBILITY に設定する必要があります。
構成	NONE、BACKWARD、BACKWARD_TRANSITIVE、FORWARD、FORWARD_TRANSITIVE、FULL、または FULL_TRANSITIVE のいずれかの値に設定する必要があります (これらの各値について詳しくは、互換性ルールに関するセクションを参照してください)。

curl 要求の例:

curl -u token:$APIKEY -X PUT $URL/rules/COMPATIBILITY -d '{"type":"COMPATIBILITY","config":"BACKWARD"}'

応答の例:

{"type":"COMPATIBILITY","config":"BACKWARD"}

グローバル・ルール構成を更新するには、少なくとも以下が必要です。

Event Streams クラスター・リソース・タイプに対する管理者役割アクセス

Activity Tracker イベントが生成され、アクションが報告されます。詳しくは、Activity Tracker イベントを参照してください。

グローバル・ルールの現行値の取得

グローバルルールの現在の値は、/rules/ {rule-type} エンドポイントに GET リクエストを発行することで取得できます（ {rule-type} は取得するグローバルルールのタイプです。現在サポートされているタイプは COMPATIBILITY のみです）。要求が成功すると、現在のルール構成が、状況コード 200 (OK) と共に応答のペイロードで返されます。

curl 要求の例:

curl -u token:$APIKEY $URL/rules/COMPATIBILITY

応答の例:

{"type":"COMPATIBILITY","config":"BACKWARD"}

グローバル・ルール構成を取得するには、少なくとも以下が必要です。

Event Streams クラスター・リソース・タイプに対するリーダー役割アクセス

スキーマごとのルールの作成

/artifacts/{schema-id}/rules エンドポイント（ {schema-id} はスキーマのID）にPOSTリクエストを送信することで、特定のスキーマにルールを適用し、設定済みのグローバルルールを上書きすることができます。リクエストの本文には、新しいルールのタイプと値を含めます（現在サポートされているタイプは「COMPATIBILITY」のみです）。成功すると、空の応答と状況コード 204 (コンテンツなし) が返されます。

curl 要求の例:

curl -u token:$APIKEY $URL/artifacts/my-schema/rules -d '{"type":"COMPATIBILITY","config":"FORWARD"}'

スキーマごとのルールを作成するには、少なくとも以下が必要です。

Event Streams クラスター・リソース・タイプに対するリーダー役割アクセス
マネージャーロールは、ルールが適用されるスキーマリソースへのアクセス権限を持つ。

Activity Tracker イベントが生成され、アクションが報告されます。詳しくは、Activity Tracker イベントを参照してください。

スキーマごとのルールの取得

特定のスキーマに適用されているルールの現在の値を取得するには、 /artifacts/{schema-id}/rules/{rule-type} エンドポイントに GET リクエストを送信します（ {schema-id} はスキーマの ID、 {rule-type} は取得するグローバルルールのタイプです。現在サポートされているタイプは COMPATIBILITY のみです）。要求が成功すると、現在のルール値が、状況コード 200 (OK) と共に応答のペイロードで返されます。

curl 要求の例:

curl -u token:$APIKEY $URL/artifacts/my-schema/rules/COMPATIBILITY

応答の例:

{"type":"COMPATIBILITY","config":"FORWARD"}

スキーマごとのルールを取得するには、少なくとも以下が必要です。

Event Streams クラスター・リソース・タイプに対するリーダー役割アクセス
ルールが適用されるスキーマ・リソースに対するリーダー役割アクセス

スキーマごとのルールの更新

特定のスキーマに適用されるルールは、 /artifacts/{schema-id}/rules/{rule-type} エンドポイントに PUT リクエストを行うことで変更できます（ {schema-id} はスキーマのID、 {rule-type} は取得するグローバルルールのタイプです。現在サポートされているタイプは COMPATIBILITY のみです）。要求が成功すると、新しく更新されたルール構成が、状況コード 200 (OK) と共に応答のペイロードで返されます。

curl 要求の例:

curl -u token:$APIKEY -X PUT $URL/artifacts/my-schema/rules/COMPATIBILITY -d '{"type":"COMPATIBILITY","config":"BACKWARD"}'

応答の例:

{"type":"COMPATIBILITY","config":"BACKWARD"}

スキーマごとのルールを更新するには、少なくとも以下が必要です。

Event Streams クラスター・リソース・タイプに対するリーダー役割アクセス
ルールが適用されるスキーマ・リソースに対する管理者役割アクセス

Activity Tracker イベントが生成され、アクションが報告されます。詳しくは、Activity Tracker イベントを参照してください。

スキーマごとのルールの削除

特定のスキーマに適用されたルールは、 /artifacts/{schema-id}/rules/{rule-type} エンドポイントに DELETE リクエストを送信することで削除されます（ {schema-id} はスキーマのID、 {rule-type} は取得するグローバルルールのタイプです。現在サポートされているタイプは COMPATIBILITY のみです）。要求が成功すると、空の応答と状況コード 204 (コンテンツなし) が返されます。

curl 要求の例:

curl -u token:$APIKEY -X DELETE $URL/artifacts/my-schema/rules/COMPATIBILITY

スキーマごとのルールを削除するには、少なくとも以下が必要です。

Event Streams クラスター・リソース・タイプに対するリーダー役割アクセス
ルールが適用されるスキーマ・リソースに対する管理者役割アクセス

Activity Tracker イベントが生成され、アクションが報告されます。詳しくは、Activity Tracker イベントを参照してください。

スキーマの新規バージョンへの互換性ルールの適用

スキーマレジストリは、スキーマの新しいバージョンを作成する際に、互換性ルールを強制的に適用することをサポートしています。必要な互換性ルールに準拠していない新規スキーマ・バージョンを作成する要求が行われると、レジストリーは要求を拒否します。以下のルールがサポートされます。

互換性ルール	テスト対象	説明
なし	該当なし	新規スキーマ・バージョンの作成時に互換性チェックは実行されません。
BACKWARD	スキーマの最新バージョン	スキーマの新規バージョンでは、スキーマの既存バージョンに存在するフィールドを省略できます。
BACKWARD_TRANSITIVE	スキーマのすべてのバージョン	スキーマの新規バージョンでは、スキーマの既存バージョンに存在しないオプション・フィールドを追加できます。
FORWARD	スキーマの最新バージョン	スキーマの新規バージョンでは、スキーマの既存バージョンに存在しないフィールドを追加できます。
FORWARD_TRANSITIVE	スキーマのすべてのバージョン	スキーマの新規バージョンでは、スキーマの既存バージョンに存在するオプション・フィールドを省略できます。
FULL	スキーマの最新バージョン	スキーマの新規バージョンでは、スキーマの既存バージョンに存在しないオプション・フィールドを追加できます。
FULL_TRANSITIVE	スキーマのすべてのバージョン	スキーマの新規バージョンでは、スキーマの既存バージョンに存在するオプション・フィールドを省略できます。

これらのルールは、以下の 2 つの適用範囲で適用できます。

グローバル適用範囲。これは、新規スキーマ・バージョンの作成時に使用されるデフォルトです。
スキーマごとのレベル。スキーマごとのレベルのルールが定義されている場合、特定のスキーマのグローバル・デフォルトはオーバーライドされます。

デフォルトでは、レジストリーのグローバル互換性ルール設定は NONE です。スキーマレベルごとのルールを定義する必要があります。そうでない場合、スキーマはグローバル設定を使用するデフォルト設定となります。

完全な API の説明

REST API の説明と例については、 Event Streams schema-registry-restを参照してください。

APIの仕様書は、 Event Streams のスキーマレジストリREST API YAMLファイルからダウンロードできます。 Swaggerファイルを表示するには、 SwaggerエディタなどのSwaggerツールを使用します。

SDK を使用してスキーマ・レジストリーにアクセスする方法について詳しくは、 Event Streams スキーマ・レジストリー REST APIを参照してください。

Terraform 上の Event Streams リソースおよびデータ・ソースについては、リソースおよびデータ・ソースを参照してください。

サードパーティ製ソフトウェアとスキーマレジストリを使用する SerDes

スキーマレジストリは、以下のサードパーティ製ソフトウェアの使用をサポートしています。 SerDes:

Confluent SerDes

スキーマ・レジストリーを使用するように Confluent SerDes を構成するには、Kafka クライアントの構成で以下の 2 つのプロパティーを指定する必要があります。

プロパティー名値

SCHEMA_REGISTRY_URL_CONFIG これは、基本認証としての資格情報およびパス /confluent を含む、スキーマ・レジストリーの URL に設定します。例えば、 $APIKEY が使用するAPIキーであり、 $HOST が 「サービス認証情報」タブの「ホスト」フィールドの kafka_http_url である場合、値は次の形式になります。 https://token:{$APIKEY}@{$HOST}/{confluent}

BASIC_AUTH_CREDENTIALS_SOURCE URL に設定します。これは、スキーマ・レジストリー URL で指定された資格情報を使用して HTTP 基本認証を使用するように SerDes に指示します。

プロパティー名	値
SCHEMA_REGISTRY_URL_CONFIG	これは、基本認証としての資格情報およびパス `/confluent` を含む、スキーマ・レジストリーの URL に設定します。例えば、 `$APIKEY` が使用するAPIキーであり、 `$HOST` が「サービス認証情報」タブの「ホスト」フィールドの `kafka_http_url` である場合、値は次の形式になります。 `https://token:{$APIKEY}@{$HOST}/{confluent}`
BASIC_AUTH_CREDENTIALS_SOURCE	`URL` に設定します。これは、スキーマ・レジストリー URL で指定された資格情報を使用して HTTP 基本認証を使用するように SerDes に指示します。

オプションで、以下のプロパティーを指定して、スキーマ選択 (サブジェクト命名方針) を制御することもできます。

プロパティー名値

VALUE_SUBJECT_NAME_STRATEGY TopicNameStrategy(デフォルト)、RecordNameStrategy、およびTopicRecordNameStrategyがサポートされます。例えば、メッセージ値のスキーマが TopicRecordNameStrategy を使用して選択されるように指定するには、以下のクライアントプロパティを使用します。 configs.put ( KafkaAvroSerializerConfig.VALUE_SUBJECT_NAME_STRATEGY, TopicRecordNameStrategy.class.getName( ));

KEY_SUBJECT_NAME_STRATEGY TopicNameStrategy (デフォルト)、RecordNameStrategy、およびTopicRecordNameStrategyがサポートされます。例については、VALUE_SUBJECT_NAME_STRATEGY を参照してください。

プロパティー名	値
VALUE_SUBJECT_NAME_STRATEGY	`TopicNameStrategy`(デフォルト)、`RecordNameStrategy`、および`TopicRecordNameStrategy`がサポートされます。例えば、メッセージ値のスキーマが `TopicRecordNameStrategy` を使用して選択されるように指定するには、以下のクライアントプロパティを使用します。 configs.put ( KafkaAvroSerializerConfig.VALUE_SUBJECT_NAME_STRATEGY, TopicRecordNameStrategy.class.getName( ));
KEY_SUBJECT_NAME_STRATEGY	`TopicNameStrategy` (デフォルト)、`RecordNameStrategy`、および`TopicRecordNameStrategy`がサポートされます。例については、VALUE_SUBJECT_NAME_STRATEGY を参照してください。

次の図は、Confluent SerDes を使用し、 Event Streams サービスに接続可能な Kafka プロデューサーを作成するために必要なプロパティの例を示しています

登録されていないスキーマを使用してメッセージが送信された場合、 SerDes は、新しいスキーマまたはスキーマのバージョンを登録するよう試みます。この動作が必要でない場合は、アプリケーションからスキーマリソースのライター権限を削除することで無効にすることができます。スキーマ・レジストリーへのアクセスの管理を参照してください。

スキーマ・ルックアップおよび登録の正規化オプションはサポートされていません。

Confluent レジストリー API を使用するツールでのスキーマ・レジストリーの使用

スキーマ・レジストリーは、Confluent Schema Registry のバージョン 7.2 によって提供される API のサブセットをサポートします。これは、Confluent Schema Registry と連携するように設計されたツールとの互換性を制限することを目的としています。 HTTP のRESTエンドポイントで、以下のパスを持つもののみが実装されています

互換性
構成
スキーマ
subjects

この互換性 API を使用するようにアプリケーションを構成するには、以下の形式でスキーマ・レジストリー・エンドポイントを指定します。

https://token:{$APIKEY}@{$HOST}/{confluent}

ここで、

$APIKEY は、 「サービス資格情報」 タブから使用する API キーです。
$HOST は、 「サービス資格情報」 タブの kafka_http_url フィールドのホストです。

サードパーティ製ツールとスキーマレジストリの併用

Schema Registryは、 kafka-avro-console-producer.sh や kafka-avro-console-consumer.sh などのサードパーティツールでテストすることができます。Confluent SerDes を使用して、スキーマへの適合性をテストすることができます。

プロデューサツールまたはコンシューマツールを実行するには、 Event Streams エンタープライズインスタンスの接続オプションで共通のプロパティが必要です。

sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="token" password="apikey";
security.protocol=SASL_SSL
sasl.mechanism=PLAIN
ssl.protocol=TLSv1.2
ssl.enabled.protocols=TLSv1.2
ssl.endpoint.identification.algorithm=HTTPS

Avro コンソールのプロデューサーおよびコンシューマー

Kafka avro コンソールのプロデューサー・ツールおよびコンシューマー・ツールを Event Streams で使用できます。クライアントプロパティを提供する必要があります。さらに、スキーマレジストリの接続方法と認証情報は、コマンドラインの --property 引数として提供する必要があります。 USER_INFOまたは URL のクレデンシャルソースを使用した2つの接続方法があります。

URL の認証情報ソースメソッドを使用して実行するには、以下のコードを使用します。

    ./kafka-avro-console-[producer|consumer] --broker-list $BOOTSTRAP_ENDPOINTS --topic schema-test --property schema.registry.url=$SCHEMA_REGISTRY_URL --property value.schema='{"type":"record","name":"myrecord","fields":[{"name":"f1","type":"string"}]}' --property basic.auth.credentials.source=URL --producer.config $CONFIG_FILE

以下の例の変数を、お客様の値に置き換えてください。

ブートストラップ・サーバーのリストとして、 IBM Cloud コンソールの Event Streams 「サービス資格情報」 タブの値を持つ BOOTSTRAP_ENDPOINTS
SCHEMA_REGISTRY_URL は IBM Cloud コンソールの Event Streams サービス資格情報タブのkafka_http_url値と、ユーザー名 tokenおよび apikey、およびパス/confluent (例えば、https://{token}:{apikey}@{kafka_http_url}/{confluent}) が付きます。
CONFIG_FILE を構成ファイルのパスに置き換えます。

USER_INFOのクレデンシャルソースメソッドを使用して実行するには、以下のコードを使用します。

    ./kafka-avro-console-[producer|consumer] --broker-list $BOOTSTRAP_ENDPOINTS --topic schema-test --property schema.registry.url=$SCHEMA_REGISTRY_URL --property value.schema='{"type":"record","name":"myrecord","fields":[{"name":"f1","type":"string"}]}' --property basic.auth.credentials.source=USER_INFO --property basic.auth.user.info=token:apikey --producer.config $CONFIG_FILE

以下の例の変数を、お客様の値に置き換えてください。

ブートストラップ・サーバーのリストとして、 IBM Cloud コンソールの Event Streams 「サービス資格情報」 タブの値を持つ BOOTSTRAP_ENDPOINTS
SCHEMA_REGISTRY_URL は IBM Cloud コンソールの Event Streams サービス資格情報タブのkafka_http_url 値とパス /confluent (例えば、https://{kafka_http_url}/{confluent}) が付きます。
CONFIG_FILE を構成ファイルのパスに置き換えます。