Avro (Java) を使用して Apache Kafka アプリケーションのスキーマを検証する

[アーティクル]
05/04/2023

このクイックスタートガイドでは、Event Hubs 用の Azure スキーマレジストリを使用して Apache Kafka アプリケーションからのイベントを検証する方法について説明します。

このユースケースの場合、Kafka プロデューサーアプリケーションでは、Azure スキーマレジストリに保存されている Avro スキーマを使用して、イベントをシリアル化し、Azure Event Hubs の Kafka トピックまたはイベントハブに発行します。 Kafka コンシューマーでは、それが使用する、Event Hubs からのイベントが逆シリアル化されます。そのために、イベントと Avro スキーマのスキーマ ID が使用されます。この ID は Azure スキーマレジストリに保存されています。

Diagram showing schema serialization/de-serialization for Kafka applications using Avro schema.

前提条件

Azure Event Hubs を初めて使用する場合は、このクイックスタートを行う前に Event Hubs の概要を参照してください。

このクイックスタートを完了するには、次の前提条件を用意しておく必要があります。

Azure サブスクリプションをお持ちでない場合は、開始する前に無料アカウントを作成してください。
開発環境で、次のコンポーネントをインストールします。
- Java Development Kit (JDK) 1.7 以降
- Maven バイナリアーカイブのダウンロードとインストール
- Git
Kafka 用 Azure スキーマレジストリリポジトリをクローンします。

イベントハブの作成

「Event Hubs 名前空間とイベントハブを作成する」のクイックスタートの手順に従って、Event Hubs 名前空間とイベントハブを作成します。次に、接続文字列を取得する方法に関するページの手順に従って、使用している Event Hubs 名前空間への接続文字列を取得します。

現在のクイックスタートで使用する次の設定をメモします。

Event Hubs 名前空間の接続文字列
イベントハブの名前

スキーマの作成

スキーマレジストリを使用してスキーマを作成する方法に関するページの手順に従って、スキーマグループとスキーマを作成します。

スキーマレジストリポータルを使用して、contoso-sg という名前のスキーマグループを作成します。シリアル化の種類として Avro を使用し、互換性モードに [なし] を使用します。

そのスキーマグループで、次のスキーマコンテンツを使用して、スキーマ名 Microsoft.Azure.Data.SchemaRegistry.example.Order を使用して新しい Avro スキーマを作成します。

{
  "namespace": "Microsoft.Azure.Data.SchemaRegistry.example",
  "type": "record",
  "name": "Order",
  "fields": [
    {
      "name": "id",
      "type": "string"
    },
    {
      "name": "amount",
      "type": "double"
    },
    {
      "name": "description",
      "type": "string"
    }
  ]
}

スキーマレジストリにアクセスするアプリケーションを登録する

Microsoft Entra ID を使用して、クライアントアプリケーションを Azure portal から Microsoft Entra テナントに登録することで、Kafka プロデューサーおよびコンシューマーアプリケーションによる Azure スキーマレジストリリソースへのアクセスを承認できます。

example-app という名前の Microsoft Entra アプリケーションの登録については、「アプリケーションを Microsoft Entra テナントに登録する」をご覧ください。

tenant.id - アプリケーションのテナント ID を設定します
client.id - アプリケーションのクライアント ID を設定します
client.secret - 認証用のクライアントシークレットを設定します

さらに、マネージド ID を使用している場合は、以下が必要になります。

use.managed.identity.credential - MSI 資格情報を使用する必要があることを示します。MSI 対応 VM に使用してください
managed.identity.clientId - 指定した場合は、指定されたクライアント ID を使用して MSI 資格情報が構築されます
managed.identity.clientId - 指定した場合は、指定されたリソース ID を使用して MSI 資格情報が構築されます

スキーマレジストリ閲覧者ロールにユーザーを追加する

名前空間レベルでスキーマレジストリ閲覧者ロールにユーザーアカウントを追加します。 スキーマレジストリ共同作成者ロールを使用することもできますが、このクイックスタートでは必要ありません。

[Event Hubs 名前空間] ページで、左側のメニューの [アクセス制御 (IAM)] を選択します。
[アクセス制御 (IAM)] ページのメニューで、[+ 追加] ->[ロールの割り当てを追加] を選択します。
[割り当ての種類] ページで、[次へ] を選択します。
[ロール] ページで、[スキーマレジストリ閲覧者 (プレビュー)] を選択し、ページの下部にある [次へ] を選択します。
[+ Select members] (+ メンバーを選択する) リンクを使用して、前の手順で作成した example-app アプリケーションをロールに追加し、[次へ] を選択します。
[確認と割り当て] ページで、[確認と割り当て] を選択します。

Kafka アプリケーションのクライアントアプリケーション構成を更新する

Kafka プロデューサーアプリケーションとコンシューマーアプリケーションのクライアント構成を、作成した Microsoft Entra アプリケーションに関連した構成とスキーマレジストリの情報を使用して更新する必要があります。

Kafka プロデューサーの構成を更新するには、azure-schema-registry-for-kafka/tree/master/java/avro/samples/kafka-producer に移動します。

Event Hubs の Kafka クイックスタートガイドに従って、src/main/resources/app.properties で Kafka アプリケーションの構成を更新します。
スキーマレジストリ関連の構成と上記で作成した Microsoft Entra アプリケーションを使用して、src/main/resources/app.properties のプロデューサーの構成の詳細を、次のように更新します。
```
schema.group=contoso-sg
schema.registry.url=https://<NAMESPACENAME>.servicebus.windows.net

 tenant.id=<>
 client.id=<>
 client.secret=<>
```
同じ手順に従って、azure-schema-registry-for-kafka/tree/master/java/avro/samples/kafka-consumer の構成も更新します。

Kafka プロデューサーアプリケーションとコンシューマーアプリケーションのどちらにも、次の Avro スキーマが使用されます。

{
  "namespace": "com.azure.schemaregistry.samples",
  "type": "record",
  "name": "Order",
  "fields": [
    {
      "name": "id",
      "type": "string"
    },
    {
      "name": "amount",
      "type": "double"
    },
    {
      "name": "description",
      "type": "string"
    }
  ]
}

Avro スキーマ検証で Kafka プロデューサーを使用する

Kafka プロデューサーアプリケーションを実行するには、azure-schema-registry-for-kafka/tree/master/java/avro/samples/kafka-producer に移動します。

プロデューサーアプリケーションを実行すると、Avro 固有のレコードまたは汎用レコードをそのアプリケーションで生成できます。特定のレコードのモードでは、まず、次の maven コマンドを使用して、プロデューサースキーマに対してクラスを生成する必要があります。
```
mvn generate-sources
```
その後に、次のコマンドを使用してプロデューサーアプリケーションを実行できます。
```
mvn clean package
mvn -e clean compile exec:java -Dexec.mainClass="com.azure.schemaregistry.samples.producer.App"
```
プロデューサーアプリケーションが正常に実行されると、プロデューサーシナリオを選択するように求められます。このクイックスタートでは、オプション 1 - produce SpecificRecords を選択します。
```
Enter case number:
1 - produce Avro SpecificRecords
2 - produce Avro GenericRecords
```

データのシリアル化と発行が正常に行われると、プロデューサーアプリケーションに次のコンソールログが表示されます。

INFO com.azure.schemaregistry.samples.producer.KafkaAvroSpecificRecord - Sent Order {"id": "ID-0", "amount": 10.0, "description": "Sample order 0"}
INFO com.azure.schemaregistry.samples.producer.KafkaAvroSpecificRecord - Sent Order {"id": "ID-1", "amount": 11.0, "description": "Sample order 1"}
INFO com.azure.schemaregistry.samples.producer.KafkaAvroSpecificRecord - Sent Order {"id": "ID-2", "amount": 12.0, "description": "Sample order 2"}

Avro スキーマ検証で Kafka コンシューマーを使用する

Kafka コンシューマーアプリケーションを実行するには、azure-schema-registry-for-kafka/tree/master/java/avro/samples/kafka-consumer に移動します。

コンシューマーアプリケーションを実行すると、Avro 固有のレコードまたは汎用レコードをそのアプリケーションで使用できます。特定のレコードのモードでは、まず、次の maven コマンドを使用して、プロデューサースキーマに対してクラスを生成する必要があります。
```
mvn generate-sources
```
その後に、次のコマンドを使用してコンシューマーアプリケーションを実行できます。
```
mvn clean package
mvn -e clean compile exec:java -Dexec.mainClass="com.azure.schemaregistry.samples.consumer.App"
```
コンシューマーアプリケーションが正常に実行されると、プロデューサーシナリオを選択するように求められます。このクイックスタートでは、オプション 1 - consume Avro SpecificRecords を選択します。
```
Enter case number:
1 - consume Avro SpecificRecords
2 - consume Avro GenericRecords
```

データの使用と逆シリアル化が正常に行われると、プロデューサーアプリケーションに次のコンソールログが表示されます。

INFO com.azure.schemaregistry.samples.consumer.KafkaAvroSpecificRecord - Order received: {"id": "ID-0", "amount": 10.0, "description": "Sample order 0"}
INFO com.azure.schemaregistry.samples.consumer.KafkaAvroSpecificRecord - Order received: {"id": "ID-1", "amount": 11.0, "description": "Sample order 1"}
INFO com.azure.schemaregistry.samples.consumer.KafkaAvroSpecificRecord - Order received: {"id": "ID-2", "amount": 12.0, "description": "Sample order 2"}

リソースをクリーンアップする

Event Hubs 名前空間を削除するか、名前空間を含むリソースグループを削除します。

Avro (Java) を使用して Apache Kafka アプリケーションのスキーマを検証する

前提条件

イベント ハブの作成

スキーマの作成

スキーマ レジストリにアクセスするアプリケーションを登録する

スキーマ レジストリ閲覧者ロールにユーザーを追加する

Kafka アプリケーションのクライアント アプリケーション構成を更新する