第9章 Service Registry を使用したスキーマの管理

Service Registry を使用すると、クライアントアプリケーションの設定からスキーマ管理のプロセスが分離されます。クライアントコードに URL を指定して、アプリケーションがレジストリーからスキーマを使用できるようにします。

たとえば、メッセージをシリアライズおよびデシリアライズするスキーマをレジストリーに保存できます。保存後、スキーマを使用するアプリケーションから参照され、アプリケーションが送受信するメッセージがこれらのスキーマと互換性を維持するようにします。

Kafka クライアントアプリケーションは実行時にスキーマを Service Registry からプッシュまたはプルできます。

スキーマは進化するので、Service Registry でルールを定義できます。たとえば、スキーマへの変更が有効で、アプリケーションによって使用される以前のバージョンとの互換性を維持するようにします。Service Registry は、変更済みのスキーマと前バージョンのスキーマを比較することで、互換性をチェックします。

Service Registry は Avro スキーマのスキーマレジストリーを完全にサポートします。Avro スキーマは、Service Registry で提供される Kafka クライアントのシリアライザー/デシリアライザー (SerDe) サービスを通じてクライアントアプリケーションによって使用されます。

プロデューサークライアントアプリケーションは、シリアライザーを使用して、特定のブローカートピックに送信するメッセージを正しいデータ形式にします。

プロデューサーが Service Registry を使用してシリアライズできるようにするには、以下を行います。

スキーマを登録したら、Kafka および Service Registry を開始するときに、スキーマにアクセスして、プロデューサーにより Kafka ブローカートピックに送信されるメッセージをフォーマットできます。

スキーマがすでに存在する場合、Service Registry に定義される互換性ルールに基づいて、REST API により新バージョンのスキーマを作成できます。バージョンは、スキーマの進化にともなう互換性チェックに使用します。アーティファクト ID およびスキーマバージョンは、スキーマを識別する一意のタプルを表します。

コンシューマークライアントアプリケーションは、デシリアライザーを使用することで、そのアプリケーションが消費するメッセージを特定のブローカートピックから正しいデータ形式にします。

コンシューマーがデシリアライズに Service Registry を使用できるようにするには、以下を実行します。

次に、消費されるメッセージに書き込まれたグローバル ID を使用して、デシリアライザーによってスキーマが取得されます。このため、受信されるメッセージにはグローバル ID およびメッセージデータが含まれる必要があります。

以下に例を示します。

# ...
[MAGIC_BYTE]
[GLOBAL_ID]
[MESSAGE DATA]

これで、Kafka および Service Registry を開始するとき、スキーマにアクセスして、Kafka ブローカートピックから受信するメッセージをフォーマットできます。

Service Registry ストラテジー は、Service Registry でメッセージスキーマが登録されるアーティファクト ID またはグローバル ID を判断するために、Kafka クライアントシリアライザー/デシリアライザーによって使用されます。

特定のトピックおよびメッセージで、以下の Java クラスの実装を使用できます。

返されるアーティファクト ID は、メッセージの キー または 値 のどちらがシリアライズされるかによって異なります。

各 ストラテジー のクラスは、io.apicurio.registry.utils.serde.strategy パッケージにまとめられています。

デフォルトのストラテジー、TopicIdStrategy は、メッセージを受信する Kafka トピックと同じ名前の Service Registry アーティファクトを検索します。

以下に例を示します。

public String artifactId(String topic, boolean isKey, T schema) {
    return String.format("%s-%s", topic, isKey ? "key" : "value");
}

使用する検索アップストラテジーは、スキーマを保存する方法と場所によって異なります。たとえば、同じ Avro メッセージタイプを持つ Kafka トピックが複数ある場合、レコード ID を使用するストラテジーを使用することがあります。

アーティファクト ID を返すストラテジー

これらのストラテジーは、ArtifactIdStrategy の実装に基づいてアーティファクト ID を返します。

グローバル ID を返すストラテジー

これらのストラテジーは、GlobalIdStrategy の実装に基づいてグローバル ID を返します。

このセクションで概説する定数を使用して、特定のクライアントの SerDe サービスおよびスキーマ検索ストラテジーを直接クライアントに設定できます。

または、プロパティーファイルまたはプロパティーインスタンスで定数を指定することもできます。

シリアライザー/デシリアライザー (SerDe) サービスの定数

public abstract class AbstractKafkaSerDe<T extends AbstractKafkaSerDe<T>> implements AutoCloseable {
      protected final Logger log = LoggerFactory.getLogger(getClass());

      public static final String REGISTRY_URL_CONFIG_PARAM = "apicurio.registry.url"; 1
      public static final String REGISTRY_CACHED_CONFIG_PARAM = "apicurio.registry.cached"; 2
      public static final String REGISTRY_ID_HANDLER_CONFIG_PARAM = "apicurio.registry.id-handler"; 3
      public static final String REGISTRY_CONFLUENT_ID_HANDLER_CONFIG_PARAM = "apicurio.registry.as-confluent"; 4

検索ストラテジーの定数

public abstract class AbstractKafkaStrategyAwareSerDe<T, S extends AbstractKafkaStrategyAwareSerDe<T, S>> extends AbstractKafkaSerDe<S> {
      public static final String REGISTRY_ARTIFACT_ID_STRATEGY_CONFIG_PARAM = "apicurio.registry.artifact-id"; 1
      public static final String REGISTRY_GLOBAL_ID_STRATEGY_CONFIG_PARAM = "apicurio.registry.global-id"; 2

コンバーターの定数

public class SchemalessConverter<T> extends AbstractKafkaSerDe<SchemalessConverter<T>> implements Converter {
      public static final String REGISTRY_CONVERTER_SERIALIZER_PARAM = "apicurio.registry.converter.serializer"; 1
      public static final String REGISTRY_CONVERTER_DESERIALIZER_PARAM = "apicurio.registry.converter.deserializer"; 2

Avro データプロバイダーの定数

public interface AvroDatumProvider<T> {
      String REGISTRY_AVRO_DATUM_PROVIDER_CONFIG_PARAM = "apicurio.registry.avro-datum-provider"; 1
      String REGISTRY_USE_SPECIFIC_AVRO_READER_CONFIG_PARAM = "apicurio.registry.use-specific-avro-reader"; 2

DefaultAvroDatumProvider (io.apicurio.registry.utils.serde.avro) 1
ReflectAvroDatumProvider (io.apicurio.registry.utils.serde.avro) 2

AMQ Streams ストレージで Service Registry をインストールする手順は、Service Registry のドキュメントを参照してください。

クラスターの設定に応じて、複数の Service Registry インスタンスをインストールできます。インスタンス数は、使用するストレージタイプと、処理する必要のあるスキーマの数によって異なります。

スキーマを Apache Avro などの適切な形式で定義したら、スキーマを Service Registry に追加できます。

スキーマは以下を使用して追加できます。

スキーマを登録するまでは、クライアントアプリケーションは Service Registry を使用できません。

curl の例

curl -X POST -H "Content-type: application/json; artifactType=AVRO" \
  -H "X-Registry-ArtifactId: prices-value" \
  --data '{ 1
      "type":"record",
      "name":"price",
      "namespace":"com.redhat",
      "fields":[{"name":"symbol","type":"string"},
      {"name":"price","type":"string"}]
    }'
  my-cluster-service-registry-myproject.example.com/artifacts -s 2

プラグインの例

<plugin>
<groupId>io.apicurio</groupId>
<artifactId>apicurio-registry-maven-plugin</artifactId>
<version>${registry.version}</version>
<executions>
  <execution>
    <phase>generate-sources</phase>
    <goals>
      <goal>register</goal>
    </goals>
    <configuration>
      <registryUrl>https//my-cluster-service-registry-myproject.example.com</registryUrl>
      <artifactType>AVRO</artifactType>
      <artifacts>
        <schema1>${project.basedir}/schemas/schema1.avsc</schema1>
      </artifacts>
    </configuration>
  </execution>
</executions>
</plugin>

(プロデューサー) クライアントによる設定例

String registryUrl_node1 = PropertiesUtil.property(clientProperties, "registry.url.node1", 1
    "https//my-cluster-service-registry-myproject.example.com");
try (RegistryService service = RegistryClient.create(registryUrl_node1)) {
    String artifactId = ApplicationImpl.INPUT_TOPIC + "-value";
    try {
        service.getArtifactMetaData(artifactId); 2
    } catch (WebApplicationException e) {
        CompletionStage <ArtifactMetaData> csa = service.createArtifact(
            ArtifactType.AVRO,
            artifactId,
            new ByteArrayInputStream(LogInput.SCHEMA$.toString().getBytes())
        );
        csa.toCompletableFuture().get();
    }
}

この手順では、Service Registry からのスキーマを使用するように Java プロデューサークライアントを設定する方法について説明します。

この手順では、Service Registry からのスキーマを使用するように Java コンシューマークライアントを設定する方法について説明します。