ウェブサイト検索

Kafka上のConfluent Schema Registryを用いてデータシリアル化とバージョン管理を効率化する

紹介

ビジネスが成長するにつれて、イベントストリーミングの場合にデータフォーマット間の互換性を維持することは非常に重要です。Apache Kafkaトピックに保存されたデータは不変であり、現在のニーズに合わせて遡って変更することはできません。これを回避することは、大量のスキーマ変更がある場合に難題となり得ます。

Apache Avroは、イベントストリーミングパイプラインのために設計されたデータシリアライズライブラリです。データのためにリッチな構造（スキーマと呼ばれる）を定義することができ、効率的な輸送と保存のためのシリアライズ機能を提供します。スキーマとそのバージョンを追跡するために、Confluent Schema Registryが作成されました。これは、スキーマの中央リポジトリとして機能し、その保存を処理し、相互互換性を保証します。これにより、スキーマのバージョンを手動で変換する方法を考えるのではなく、データに集中することができます。

このチュートリアルでは、Dockerを使用してConfluent Schema Registryをデプロイし、これまでのシリーズのチュートリアルで作成したKafkaプロデューサーとコンシューマーを拡張します。定義したスキーマに準拠するオブジェクトを作成および消費するようにそれらを再構築します。また、スキーマを修正し、以前のバージョンに準拠するデータを壊すことなく進化させる方法を学びます。

前提条件

このチュートリアルに従うためには、次のものが必要です:

  • あなたのマシンにDockerがインストールされている必要があります。Ubuntuの場合は、「UbuntuにDockerをインストールして使用する方法」を訪れてください。このチュートリアルのステップ1とステップ2だけを完了すれば大丈夫です。
  • あなたのマシンにDocker Composeがインストールされています。Ubuntuの場合は、「Docker ComposeをUbuntuにインストールして使用する方法」のステップ1とステップ2に従ってください。
  • CLIガイドに従ってデータをソースするように設定されたKafka プロデューサーを持つJavaプロジェクト。
  • CLIガイドに従ってデータを受信するように設定されたKafka コンシューマを持つJavaプロジェクト。
  • jqがあなたのマシンにインストールされています。全体の概要については、How To Transform JSON Data with jqの記事をご覧ください。
  • Kafkaについての理解、トピック、プロデューサー、およびコンシューマーを含みます。詳細については、「Kafkaの紹介」をご覧ください。

ステップ1 - Docker Composeを使用してスキーマレジストリを実行する

このセクションでは、Docker Composeを使用してConfluent Schema Registryを実行する方法を学びます。KRaftを使用してスタンドアロンで動作するKafkaとは異なり、Schema Registryは機能するためにZooKeeperインスタンスを必要とします。

前提条件の一部として、あなたはローカルマシンにKafkaをsystemdサービスとしてデプロイしました。このステップでは、Docker Composeを使用してKafkaノードをデプロイします。まず、次のコマンドを実行してサービスを停止する必要があります:

sudo systemctl stop kafka

これで、schema-registry-compose.yamlという名前のファイルにDocker Composeの設定を定義します。次のコマンドを実行して作成し、編集のために開きます:

nano schema-registry-compose.yaml

次の行を追加してください:

version: '3'

services:
  zookeeper:
    image: confluentinc/cp-zookeeper:latest
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000

  kafka:
    image: confluentinc/cp-kafka:latest
    depends_on:
      - zookeeper
    ports:
      - 9092:9092
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

  kafka-schema-registry:
    image: confluentinc/cp-schema-registry
    hostname: kafka-schema-registry
    container_name: kafka-schema-registry
    depends_on:
      - zookeeper
      - kafka
    ports:
      - "8081:8081"
    environment:
      SCHEMA_REGISTRY_HOST_NAME: kafka-schema-registry
      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: 'PLAINTEXT://kafka:29092'
      SCHEMA_REGISTRY_LISTENERS: http://0.0.0.0:8081

ここでは、3つのサービス zookeeper、kafka、および kafka-schema-registry を定義します。これらの3つのサービスはすべて、Confluent が提供する最新の Docker イメージを使用しています。ZooKeeper サービスはポート 2181 で公開され、kafka は 9092 で公開されます。kafka に対しては、environment の下に ZooKeeper アドレスを設定し、スキーマレジストリが Kafka に直接接続するために使用する追加のリスナー 29092 を指定します。kafka サービスは、最初に zookeeper が起動するのを待つ必要があることを指定します。

次に、kafka-schema-registryをポート8081で公開し、environmentの下にあるkafkaサービスへの接続のためのアドレスを渡します。また、Schema RegistryはZooKeeperとKafkaが初期化を完了した時のみ起動するように指定します。

ファイルを保存して閉じたら、次のコマンドを実行してバックグラウンドでサービスを起動します:

docker-compose -f schema-registry-compose.yaml up -d

出力の最後は以下の行に似たものになるでしょう：

...
Creating root_zookeeper_1 ... done
Creating root_kafka_1     ... done
Creating kafka-schema-registry ... done

実行中のコンテナを一覧表示できます:

docker ps

出力は次のようになります:

CONTAINER ID   IMAGE                              COMMAND                  CREATED          STATUS          PORTS                                       NAMES
6783568a74c8   confluentinc/cp-schema-registry    "/etc/confluent/dock…"   19 seconds ago   Up 17 seconds   0.0.0.0:8081->8081/tcp, :::8081->8081/tcp   kafka-schema-registry
6367df4b55f7   confluentinc/cp-kafka:latest       "/etc/confluent/dock…"   19 seconds ago   Up 18 seconds   0.0.0.0:9092->9092/tcp, :::9092->9092/tcp   root_kafka_1
a5f5b09984e0   confluentinc/cp-zookeeper:latest   "/etc/confluent/dock…"   19 seconds ago   Up 19 seconds   2181/tcp, 2888/tcp, 3888/tcp                root_zookeeper_1

このステップでは、Docker Composeを使用して、ZooKeeperとKafkaとともにSchema Registryインスタンスをデプロイしました。これから、JavaプロジェクトでAvroスキーマを作成し、使用する方法を学びます。

ステップ2 - Avroスキーマの使用

このセクションでは、プロジェクトにAvroを追加し、関連する依存関係も追加します。Avroスキーマを定義し、定義された型のためにJavaクラスを自動生成する方法を学びます。そして、Avroスキーマをスキーマレジストリに追加します。

プロジェクトにAvroを追加する

まず、org.apache.avro依存関係を追加します。ブラウザでJavaクライアントのMavenリポジトリページに移動し、利用可能な最新バージョンを選択して、Maven用の提供されたXMLスニペットをコピーします。執筆時点で、Javaクライアントライブラリの最新バージョンは1.11.3でした。

pom.xml に依存関係をプロジェクトのルートに追加します。編集するには、次のコマンドを実行します:

nano pom.xml

<dependencies> セクションを見つけて、依存関係の定義を追加します:

...
<!-- https://mvnrepository.com/artifact/org.apache.avro/avro -->
<dependency>
  <groupId>org.apache.avro</groupId>
  <artifactId>avro</artifactId>
  <version>1.11.3</version>
</dependency>
...

これにより、Avroがあなたのプロジェクトで利用できるようになります。ただし、AvroスキーマからJavaクラスを生成できるようにするには、同様にMavenリポジトリからavro-maven-pluginを追加する必要があります。

依存関係を定義したら、プラグインがソースを生成することを確認する必要があります。pom.xml の <build> セクションを見つけて、強調表示された行を追加してください:

...
<plugin>
  <groupId>org.apache.avro</groupId>
  <artifactId>avro-maven-plugin</artifactId>
  <version>1.11.3</version>
  <executions>
    <execution>
      <phase>generate-sources</phase>
      <goals>
        <goal>schema</goal>
      </goals>
      <configuration>
        <sourceDirectory>${project.basedir}/src/main/java/com/dokafka/</sourceDirectory>
        <outputDirectory>${project.basedir}/src/main/java/</outputDirectory>
      </configuration>
    </execution>
  </executions>
</plugin>
...

ここでは、avro-maven-pluginを設定して、/src/main/java/com/dokafkaのスキーマに基づいてJavaソースを生成し、それを/src/main/javaの下に配置します。既存の<plugins>セクションを<pluginManagement>の下で変更しないことに注意してください。

終わったら、ファイルを保存して閉じてください。

プロジェクトをビルドして、すべてが正しく設定されていることを確認してください。

mvn package

出力の最後はこのようになります:

[INFO] ------------------------------------------------------------------------
[INFO] BUILD SUCCESS
[INFO] ------------------------------------------------------------------------
[INFO] Total time:  5.759 s
[INFO] Finished at: 2024-04-01T13:38:31Z
[INFO] ------------------------------------------------------------------------

Avroスキーマの定義

現在、ある時点での温度測定を説明する TempMeasurement というスキーマを作成します。これは、前提条件として作成した ProducerDemo および ConsumerDemo クラスの隣に保存します。次のコマンドを実行して、作成し、編集のために開いてください:

nano src/main/java/com/dokafka/TempMeasurement.avsc

次の行を追加してください:

{
  "namespace": "com.dokafka",
  "name": "TempMeasurement",
  "type": "record",
  "fields": [
      {
          "name": "measuredValue",
          "type": "double"
      },
      {
          "name": "measurerName",
          "type": "string"
      }
  ]
}

Avro スキーマファイルは JSON で記述され、そのファイル拡張子は .avsc です。まず、スキーマの namespace を指定します。これは自動生成される Java クラスのネームスペースにもなります。次に、name を TempMeasurement に設定し、スキーマタイプとして record を指定します。これは Avro オブジェクトであることを示しています。

次に、あなたはスキーマのfieldsを指定します。それはそれぞれdouble型のmeasuredValueとstring型のmeasurerNameと呼ばれます。Avroは他にもint、long、float、boolean、およびbytesのようなタイプをサポートしています。

ファイルを保存して閉じたら、プロジェクトをビルドします:

mvn package

では、src/main/java/com/dokafkaの下にあるファイルをリストアップします:

ls src/main/java/com/dokafka

あなたは TempMeasurement クラスが作成されたことを目にするでしょう:

ConsumerDemo.java  ProducerDemo.java  TempMeasurement.avsc  TempMeasurement.java

このクラスは、TempMeasurement オブジェクトをインスタンス化し、シリアライズおよびデシリアライズするためのコードを保持しています。

スキーマレジストリにスキーマを保存する

スキーマを定義したので、次のコマンドを実行してそれをスキーマレジストリに追加できます。

curl -X POST -H "Content-Type: application/vnd.schemaregistry.v1+json" \
    -d "$(jq -R -s '{"schema": .}' < src/main/java/com/dokafka/TempMeasurement.avsc)" \
    http://localhost:8081/subjects/TempMeasurement/versions

スキーマレジストリはhttp://localhost:8081で公開されており、HTTPを通じてアクセスできます。このコマンドでは、最初にHTTPメソッドをPOSTに設定し、レジストリが受け入れる適切なContent-Typeを指定します。

リクエストボディを-dに渡し、jqを使用して、スキーマレジストリが受け入れる形式でschemaというフィールドにスキーマの内容をラップします。最後に、リクエストをsubjects/TempMeasurement/versionsに向け、新しいスキーマの呼び方を指定します。

出力は次の通りです:

{"id":1}

スキーマレジストリはリクエストを受け付け、IDを1として割り当てました。

利用可能なスキーマの一覧を表示するには、次のコマンドを実行してください:

curl -X GET http://localhost:8081/subjects

利用可能なのは一つだけです。

["TempMeasurement"]

このステップでは、必要な依存関係をMavenプロジェクトに追加し、定義したAvroスキーマのコード自動生成をセットアップしました。これから、スキーマに基づいてデータを生成し、消費することに取り組みます。

ステップ3 - スキーマに従った生成と消費

Confluentは、kafka-avro-serializerと呼ばれるKafka用のAvroシリアライザライブラリを提供しています。このステップでは、それをプロジェクトに追加し、プロデューサーとコンシューマーを構成してTempMeasurementオブジェクトを送信します。

必要な依存関係を追加する

プロジェクトにライブラリを追加するには、Mavenリポジトリに移動し、執筆時点での最新のバージョンである7.6.0のXML依存関係定義をコピーします。その後、編集のためにpom.xmlを開きます:

nano pom.xml

このkafka-avro-serializerパッケージはConfluentのMavenリポジトリでホストされているため、ハイライトされた行を追加して定義する必要があります。

...
<repositories>
  <repository>
    <id>confluent</id>
    <url>http://packages.confluent.io/maven/</url>
  </repository>
</repositories>
...

同様に、ライブラリ定義を<dependencies>セクションに追加します:

...
<!-- https://mvnrepository.com/artifact/io.confluent/kafka-avro-serializer -->
<dependency>
  <groupId>io.confluent</groupId>
  <artifactId>kafka-avro-serializer</artifactId>
  <version>7.6.0</version>
</dependency>
...

Kafkaクライアントがスキーマレジストリと通信できるようにするには、kafka-schema-registry-clientを依存関係として追加する必要があります。そのため、Mavenリポジトリに移動し、最新の利用可能なバージョンの定義をpom.xmlに挿入してください。

作業が完了したら、ファイルを保存して閉じてください。これで、MavenプロジェクトでKafkaAvroSerializerとKafkaAvroDeserializerクラスを使用できるようになります。

ProducerDemoとConsumerDemoを更新中

次に、ProducerDemoクラスを改修して、スキーマレジストリに接続し、トピック上でTempMeasurement型のオブジェクトを生成します。編集するには、次のコマンドを実行して開いてください:

nano src/main/java/com/dokafka/ProducerDemo.java

KafkaAvroSerializer と KafkaAvroSerializerConfig をインポートして使用できるようにします:

...
import org.apache.kafka.common.serialization.StringSerializer;
import io.confluent.kafka.serializers.KafkaAvroSerializer;
import io.confluent.kafka.serializers.KafkaAvroSerializerConfig;
...

次に、mainメソッドの最初の部分を次のように修正します:

...
String topicName = "java_demo_avro";

Properties properties = new Properties();
properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class.getName());
properties.setProperty(KafkaAvroSerializerConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://localhost:8081");

KafkaProducer<String, TempMeasurement> producer = new KafkaProducer<>(properties);

ProducerRecord<String, TempMeasurement> producerRecord =
  new ProducerRecord<>(topicName, new TempMeasurement(Double.parseDouble(args[0]), args[1]));
...

まず、トピック名をjava_demo_avroに設定します。Kafkaは、既に存在しない場合にはトピックを作成します。次に、値のシリアライザークラスをStringSerializerからKafkaAvroSerializerに切り替え、スキーマレジストリのアドレスを指定するSCHEMA_REGISTRY_URL_CONFIGパラメータを設定します。

あなたは、以前のString値の定義をTempMeasurementに置き換えます。producerRecordには、最初の2つのコマンドライン引数から取得した2つのパラメータ（measurementValueとmeasurerName）を持つTempMeasurementの新しいインスタンスを渡します。

作業が終わったら、ファイルを保存して閉じてください。それから、run-producer.sh スクリプトに2つの引数を渡せるように、編集のために開く必要があります:

nano run-producer.sh

コマンドにハイライトされたパラメータを追加してください:

...
java -cp target/dokafka-1.0-SNAPSHOT.jar:target/lib/* com.dokafka.ProducerDemo $1 $2

ファイルを保存して閉じたら、次のコマンドを実行してTempMeasurementを生成します:

./run-producer.sh 100 sammy

このコマンドでは、measurementValueとして100を、measurerNameとしてsammyを渡します。

あなたが今生産したAvroオブジェクトを受け取るためには、ConsumerDemoを同様に修正する必要があります。ファイルを編集のために開いてください:

nano src/main/java/com/dokafka/ConsumerDemo.java

KafkaAvroDeserializer と KafkaAvroDeserializerConfig をインポートして、それらを参照できるようにします:

...
import org.apache.kafka.common.serialization.StringDeserializer;
import io.confluent.kafka.serializers.KafkaAvroDeserializer;
import io.confluent.kafka.serializers.KafkaAvroDeserializerConfig;
...

次に、mainメソッドを次のように変更します。

...
String topic = "java_demo_avro";
String groupId = "group1";

Properties properties = new Properties();
properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
properties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
properties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KafkaAvroDeserializer.class.getName());
properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId);
properties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
properties.setProperty(KafkaAvroDeserializerConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://localhost:8081");
properties.setProperty(KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG, "true");

final KafkaConsumer<String, TempMeasurement> consumer = new KafkaConsumer<>(properties);
consumer.subscribe(Collections.singletonList(topic));

try {
  while (true) {
    ConsumerRecords<String, TempMeasurement> records = consumer.poll(Duration.ofMillis(100));

    for (ConsumerRecord<String, TempMeasurement> record: records) {
      log.info(String.format("measuredValue: %s, measurerName: %s\n",
        record.value().getMeasuredValue(),
        record.value().getMeasurerName()));
    }
  }
} catch (Exception e) {
  log.error("An error occurred", e);
} finally {
  consumer.close();
}
...

ConsumerDemoと同様に、topicを更新し、SCHEMA_REGISTRY_URL_CONFIGパラメータにスキーマレジストリのアドレスを設定します。また、デシリアライザをKafkaAvroDeserializerに更新します。SPECIFIC_AVRO_READER_CONFIGパラメータをtrueに設定することで、デシリアライザに実際のTempMeasurementオブジェクトを返すように指示します。そうでない場合、デシリアライザはすべてのフィールドを含むAvroのGenericRecordを返しますが、これは厳密に型付けされていません。

その後、TempMeasurementを値の型としてコード全体に伝播させます。forループ内で、measurementTempとmeasurementValueを出力するようにログメソッドの呼び出しを修正します。

スキーマレジストリの統合のおかげで、プロデューサーはオブジェクトをトピックに送信する際にオブジェクトスキーマをオブジェクトと一緒にバンドルしません。代わりに、スキーマの識別子と一緒にオブジェクトを送信します。コンシューマーはそのスキーマをスキーマレジストリから完全に取得し、次にデシリアライズします。

作業が完了したらファイルを保存して閉じ、その後コンシューマーを実行します:

./run-consumer.sh

出力の終わりはこのようになります:

...
[main] INFO com.dokafka.ConsumerDemo - measuredValue: 100.0, measurerName: sammy

あなたのKafkaコンシューマーは、ログメッセージによって証明されるように、Avroメッセージを正常にデシリアライズしました。

このステップでは、プロデューサーとコンシューマークラスをAvroオブジェクトを使用するように更新しました。次のステップでは、スキーマを更新し、それらの互換性をスキーマレジストリで追跡する方法を学びます。

ステップ4 - スキーマの進化と互換性

このステップでは、既存のスキーマを更新する方法と、これらの変更が既存のバージョンやクライアントとの互換性にどのように影響するかを学びます。

スキーマとそのバージョンを時間を通じて保存することに加えて、スキーマレジストリはスキーマの進化を可能にするために重要です。スキーマはプロジェクトの存続期間中に変更することができますが、一度生成されたデータは変更できません。

スキーマレジストリは、スキーマバージョン間の互換性を管理し、消費者が内部のスキーマバージョンに従ってできるだけ多くのデータを解析できるようにします。これにより、プロデューサーと消費者は、異なるコードベースに存在できるため、正確なスキーマバージョンに関して非同期でいることが可能です。

スキーマレジストリが提供する主な互換性戦略は次のとおりです:

  • BACKWARDは、新しいスキーマを使用する消費者が以前のバージョンに基づいてデータを読み取ることができることを保証します。
  • FORWARD、これは新しいスキーマを使用する消費者が新しいスキーマに基づいてデータを読み取ることができることを意味します（以前のスキーマについては保証がありません）
  • FULLは、以前の2つの戦略を組み合わせたものです。
  • NONEは、互換性チェックが無効であることを意味します

最初の3つの戦略には、BACKWARD_TRANSITIVEのような推移的な対応物もあり、新しいスキーマはすべての以前のバージョンと互換性がある必要があり、直前のバージョンだけではありません。デフォルトの戦略はBACKWARDです。

これからTempMeasurementスキーマを修正し、測定の日付を保存するためのフィールドを追加します。次のコマンドを実行してスキーマを編集モードで開いてください:

nano src/main/java/com/dokafka/TempMeasurement.avsc

これのように修正してください:

{
  "namespace": "com.dokafka",
  "name": "TempMeasurement",
  "type": "record",
  "fields": [
      {
          "name": "measuredValue",
          "type": "double"
      },
      {
          "name": "measurerName",
          "type": "string"
      },
      {
          "name": "measurementDate",
          "type": "string"
      }
  ]
}

新しいフィールドmeasurementDateを定義しました。これは測定の日付をテキスト形式で保存します。作業が完了したらファイルを保存して閉じてください。

Schema Registryに新しいバージョンのスキーマを作成するには、以下のコマンドを実行してください:

curl -X POST -H "Content-Type: application/vnd.schemaregistry.v1+json" \
    -d "$(jq -R -s '{"schema": .}' < src/main/java/com/dokafka/TempMeasurement.avsc)" \
    http://localhost:8081/subjects/TempMeasurement/versions

次の出力が得られます。エラーの詳細が記載されています:

{
  "error_code": 409,
  "message": "Schema being registered is incompatible with an earlier schema for subject \"TempMeasurement\", details: [{errorType:'READER_FIELD_MISSING_DEFAULT_VALUE', description:'The field 'measurementDate' at path '/fields/2' in the new schema has no default value and is missing in the old schema', additionalInfo:'measurementDate'}, {oldSchemaVersion: 1}, {oldSchema: '{\"type\":\"record\",\"name\":\"TempMeasurement\",\"namespace\":\"com.dokafka\",\"fields\":[{\"name\":\"measuredValue\",\"type\":\"double\"},{\"name\":\"measurerName\",\"type\":\"string\"}]}'}, {validateFields: 'false', compatibility: 'BACKWARD'}]"
}

エラーは、新しいスキーマが前のスキーマと後方互換性がないと述べています。これは、measurementDateフィールドが新しいものであり、デフォルト値を持っていないためです。この動作により、新しいスキーマを持つ消費者は、古いスキーマを使用して作成されたデータを読み取ることができるようになります。その場合、存在しないフィールドのデフォルト値がないと逆シリアル化は失敗します。

ファイルを開いて、ハイライトされた行を追加してください。

{
  "namespace": "com.dokafka",
  "name": "TempMeasurement",
  "type": "record",
  "fields": [
      {
          "name": "measuredValue",
          "type": "double"
      },
      {
          "name": "measurerName",
          "type": "string"
      },
      {
          "name": "measurementDate",
          "type": "string",
          "default": ""
      }
  ]
}

measurementDate は現在、デフォルト値として空の文字列を持つようになります。ファイルを保存して閉じた後、スキーマを再度送信してみてください。出力は次のようになります:

{"id":2}

スキーマレジストリは、最初のスキーマとの下位互換性があるため、スキーマの2番目のバージョンを受け入れ、IDとして2を割り当てました。最初のバージョンを取得するには、次のコマンドを実行します:

curl -X GET http://localhost:8081/subjects/TempMeasurement/versions/1

出力は次の通りです:

{
  "subject": "TempMeasurement",
  "version": 1,
  "id": 1,
  "schema": "{\"type\":\"record\",\"name\":\"TempMeasurement\",\"namespace\":\"com.dokafka\",\"fields\":[{\"name\":\"measuredValue\",\"type\":\"double\"},{\"name\":\"measurerName\",\"type\":\"string\"}]}"
}

特定のIDを省略することで、スキーマのすべてのバージョンをリストすることもできます:

curl -X GET http://localhost:8081/subjects/TempMeasurement/versions

あなたは二つあることがわかるでしょう。

[1, 2]

互換性戦略を変更するには、次のコマンドを実行できます:

curl -X PUT -H "Content-Type: application/vnd.schemaregistry.v1+json" \
       --data '{"compatibility": "BACKWARD_TRANSITIVE"}' \
       http://localhost:8081/config/TempMeasurements

このコマンドは TempMeasurements の戦略を BACKWARD_TRANSITIVE に設定します。エンドポイントが config であり、subjects ではないことに注意してください。出力は次のとおりです:

{"compatibility":"BACKWARD_TRANSITIVE"}

開始したDocker Composeリソースを削除するには、次のコマンドを実行してください。

docker-compose -f schema-registry-compose.yaml down

このステップでは、BACKWARD 互換性戦略に従って TempMeasurement スキーマを修正し、スキーマレジストリに公開しました。

結論

この記事では、ProducerDemo と ConsumerDemo クラスを拡張して、Apache Avro でシリアライズされた TempMeasurement オブジェクトを生成および消費する方法について説明しました。スキーマストレージと進化のためのスキーマレジストリの利用方法を学び、Kafka クライアントをそれに接続しました。

著者は「Write for DOnations」プログラムの一環として、寄付を受けるためにApache Software Foundationを選びました。

関連記事: