The English version of quarkus.io is the official project site. Translated sites are community supported on a best-effort basis.

Schema RegistryとAvroと共にApache Kafkaを使用

このガイドでは、Quarkus アプリケーションが Apache Kafka、 Avro でシリアライズされたレコードを使用し、スキーマレジストリー (Confluent スキーマレジストリー または Apicurio レジストリー) に接続する方法を説明しています。

特に Kafka や Kafka in Quarkus を使い慣れていない場合は、最初に リアクティブメッセージングでの Apache Kafka の使用 ガイドを確認することをお勧めします。

前提条件

このガイドを完成させるには、以下が必要です:

  • ざっと 30 minutes

  • IDE

  • JDK 17+がインストールされ、 JAVA_HOME が適切に設定されていること

  • Apache Maven 3.9.6

  • Docker と Docker Compose、または Podman 、および Docker Compose

  • 使用したい場合は、 Quarkus CLI

  • ネイティブ実行可能ファイルをビルドしたい場合、MandrelまたはGraalVM(あるいはネイティブなコンテナビルドを使用する場合はDocker)をインストールし、 適切に設定していること

アーキテクチャ

このガイドでは、REST リソース、つまりムービー DTO を消費して Kafka トピックに配置する MovieResource を実装します。

次に、同じトピックからメッセージを消費および収集するコンシューマーを実装します。収集されたメッセージは、 Server-Sent Events を介して別のリソースである ConsumedMovieResource によって公開されます。

Movies は、Avro を使用してシリアライズおよびデシリアライズされます。Movie を記述するスキーマは、Apicurio Registry に保存されます。Confluent Avro serde および Confluent Schema Registry を使用している場合も同じ概念が適用されます。

ソリューション

次の章で紹介する手順に沿って、ステップを踏んでアプリを作成することをお勧めします。ただし、完成した例にそのまま進んでも構いません。

Git リポジトリーのクローンを作成: git clonehttps://github.com/quarkusio/quarkus-quickstarts.git、または アーカイブ をダウンロードします。

ソリューションは kafka-avro-schema-quickstart ディレクトリ にあります。

Maven プロジェクトの作成

まず、新しいプロジェクトが必要です。以下のコマンドで新規プロジェクトを作成します。

コマンドラインインタフェース
quarkus create app org.acme:kafka-avro-schema-quickstart \
    --extension='rest-jackson,messaging-kafka,apicurio-registry-avro' \
    --no-code
cd kafka-avro-schema-quickstart

Gradleプロジェクトを作成するには、 --gradle または --gradle-kotlin-dsl オプションを追加します。

Quarkus CLIのインストールと使用方法の詳細については、 Quarkus CLI ガイドを参照してください。

Maven
mvn io.quarkus.platform:quarkus-maven-plugin:3.10.1:create \
    -DprojectGroupId=org.acme \
    -DprojectArtifactId=kafka-avro-schema-quickstart \
    -Dextensions='rest-jackson,messaging-kafka,apicurio-registry-avro' \
    -DnoCode
cd kafka-avro-schema-quickstart

Gradleプロジェクトを作成するには、 -DbuildTool=gradle または -DbuildTool=gradle-kotlin-dsl オプションを追加します。

Windowsユーザーの場合:

  • cmdを使用する場合、(バックスラッシュ \ を使用せず、すべてを同じ行に書かないでください)。

  • Powershellを使用する場合は、 -D パラメータを二重引用符で囲んでください。例: "-DprojectArtifactId=kafka-avro-schema-quickstart"

Confluent Schema Registry を使用する場合、quarkus-apicurio-registry-avro エクステンションは必要ありません。代わりに、quarkus-confluent-registry-avro エクステンションと、さらにいくつかの依存関係が必要です。詳しくは、 Confluent Schema Registryの使用 を参照下さい。

Avro スキーマ

Apache Avro はデータシリアル化システムです。データ構造はスキーマを使用して記述されます。最初に Movie 構造を記述するスキーマを作成する必要があります。レコードのスキーマ (Kafka メッセージ) を使用して、src/main/avro/movie.avsc というファイルを作成します。

{
  "namespace": "org.acme.kafka.quarkus",
  "type": "record",
  "name": "Movie",
  "fields": [
    {
      "name": "title",
      "type": "string"
    },
    {
      "name": "year",
      "type": "int"
    }
  ]
}

次のコマンドでプロジェクトをビルドする場合:

コマンドラインインタフェース
quarkus build
Maven
./mvnw install
Gradle
./gradlew build

movies.avsc は、target/generated-sources/avsc ディレクトリーに配置された Movie.java ファイルにコンパイルされます。

Avro 構文とサポートされているタイプの詳細については、 Avro 仕様 を参照してください。

Quarkus を使用すると、Avro スキーマを処理するために特定の Maven プラグインを使用する必要はありません。これはすべて、quarkus-avro エクステンションによって行われます。

次のコマンドでプロジェクトを実行する場合:

コマンドラインインタフェース
quarkus dev
Maven
./mvnw quarkus:dev
Gradle
./gradlew --console=plain quarkusDev

スキーマファイルに加えた変更は、生成された Java ファイルに自動的に適用されます。

Movie プロデューサー

スキーマを定義したら、MovieResource を実装できます。

MovieResource を開き、Movie DTO の Emitter を挿入し、Movie を消費する @POST メソッドを実装して Emitter 経由で送信してください。

package org.acme.kafka;

import org.acme.kafka.quarkus.Movie;
import org.eclipse.microprofile.reactive.messaging.Channel;
import org.eclipse.microprofile.reactive.messaging.Emitter;
import org.jboss.logging.Logger;

import jakarta.ws.rs.POST;
import jakarta.ws.rs.Path;
import jakarta.ws.rs.core.Response;

@Path("/movies")
public class MovieResource {
    private static final Logger LOGGER = Logger.getLogger(MovieResource.class);

    @Channel("movies")
    Emitter<Movie> emitter;

    @POST
    public Response enqueueMovie(Movie movie) {
        LOGGER.infof("Sending movie %s to Kafka", movie.getTitle());
        emitter.send(movie);
        return Response.accepted().build();
    }

}

ここで、movies チャネル (Emitter はこのチャネルに出力) を Kafka トピックに マップ する必要があります。これを実現するには、application.properties ファイルを編集し、次のコンテンツを追加します。

# set the connector for the outgoing channel to `smallrye-kafka`
mp.messaging.outgoing.movies.connector=smallrye-kafka

# set the topic name for the channel to `movies`
mp.messaging.outgoing.movies.topic=movies

# automatically register the schema with the registry, if not present
mp.messaging.outgoing.movies.apicurio.registry.auto-register=true

value.serializer を定義しなかったことにお気づきかもしれません。これは、@Channel 宣言、Movie タイプの構造、Apicurio Registry ライブラリーの存在に基づき、io.apicurio.registry.serde.avro.AvroKafkaSerializer が適切であることを、Quarkus が 自動検出 できるためです。それでも、apicurio.registry.auto-register プロパティーは定義する必要があります。

Confluent Schema Registry を使用する場合は、value.serializer も設定する必要はありません。同様に、自動的に検出されます。apicurio.registry.auto-register の Confluent Schema Registry アナログは、auto.register.schemas と呼ばれます。デフォルトは true であるため、この例で設定する必要はありません。自動スキーマ登録を無効にする場合は、明示的に false に設定できます。

Movie コンシューマー

Movie データを含むレコードを Kafka に書き込むことができます。そのデータは Avro を使用してシリアライズされます。では、それらのコンシューマーを実装してみましょう。

movies-from-kafka チャネルからの Movie メッセージを消費し、Server-Sent Event を介して公開する ConsumedMovieResource を作成します。

package org.acme.kafka;

import jakarta.enterprise.context.ApplicationScoped;
import jakarta.ws.rs.GET;
import jakarta.ws.rs.Path;
import jakarta.ws.rs.Produces;
import jakarta.ws.rs.core.MediaType;

import org.acme.kafka.quarkus.Movie;
import org.eclipse.microprofile.reactive.messaging.Channel;
import org.jboss.resteasy.reactive.RestStreamElementType;

import io.smallrye.mutiny.Multi;

@ApplicationScoped
@Path("/consumed-movies")
public class ConsumedMovieResource {

    @Channel("movies-from-kafka")
    Multi<Movie> movies;

    @GET
    @Produces(MediaType.SERVER_SENT_EVENTS)
    @RestStreamElementType(MediaType.TEXT_PLAIN)
    public Multi<String> stream() {
        return movies.map(movie -> String.format("'%s' from %s", movie.getTitle(), movie.getYear()));
    }
}

アプリケーションのコードの最後のビットは、application.propertiesmovies-from-kafka チャネルの設定です。

# set the connector for the incoming channel to `smallrye-kafka`
mp.messaging.incoming.movies-from-kafka.connector=smallrye-kafka

# set the topic name for the channel to `movies`
mp.messaging.incoming.movies-from-kafka.topic=movies

# disable auto-commit, Reactive Messaging handles it itself
mp.messaging.incoming.movies-from-kafka.enable.auto.commit=false

mp.messaging.incoming.movies-from-kafka.auto.offset.reset=earliest

value.deserializer を定義しなかったことにお気づきかもしれません。これは、@Channel 宣言、Movie タイプの構造、Apicurio Registry ライブラリーの存在に基づき、io.apicurio.registry.serde.avro.AvroKafkaDeserializer が適切であることを、Quarkus が 自動検出 できるためです。apicurio.registry.use-specific-avro-reader プロパティーも自動で設定されるため、定義する必要はありません。

Confluent Schema Registry を使用する場合は、value.deserializerspecific.avro.reader を設定する必要はありません。どちらも自動的に検出されます。

アプリケーションの実行

アプリケーションを開発モードで実行します。

コマンドラインインタフェース
quarkus dev
Maven
./mvnw quarkus:dev
Gradle
./gradlew --console=plain quarkusDev

Dev Services により、Kafka ブローカーと Apicurio Registry インスタンスは自動的に開始されます。詳細については、Dev Services for Kafka および Dev Services for Apicurio Registry を参照してください。

You might have noticed that we didn’t configure the schema registry URL anywhere. This is because Dev Services for Apicurio Registry configures all Kafka channels in Quarkus Messaging to use the automatically started registry instance.

ネイティブ API に加えて、Apicurio Registry は Confluent Schema Registry と API 互換のエンドポイントも公開します。したがって、この自動設定は、Apicurio Registry serde と Confluent Schema Registry serde の両方で機能します。

ただし、Confluent Schema Registry 自体の実行に対する Dev Services のサポートはないことに注意してください。Confluent Schema Registry の実行インスタンスを使用する場合は、その URL を Kafka ブローカーの URL と併せて設定します。

kafka.bootstrap.servers=PLAINTEXT://localhost:9092
mp.messaging.connector.smallrye-kafka.schema.registry.url=http://localhost:8081

2 番目のターミナルで、ConsumedMovieResource リソースを curl でクエリーします。

curl -N http://localhost:8080/consumed-movies

3番目のターミナルで、いくつかの movie をポストします。

curl --header "Content-Type: application/json" \
  --request POST \
  --data '{"title":"The Shawshank Redemption","year":1994}' \
  http://localhost:8080/movies

curl --header "Content-Type: application/json" \
  --request POST \
  --data '{"title":"The Godfather","year":1972}' \
  http://localhost:8080/movies

curl --header "Content-Type: application/json" \
  --request POST \
  --data '{"title":"The Dark Knight","year":2008}' \
  http://localhost:8080/movies

curl --header "Content-Type: application/json" \
  --request POST \
  --data '{"title":"12 Angry Men","year":1957}' \
  http://localhost:8080/movies

2 番目のターミナルに表示されているものを確認します。次の行に何かが表示されているはずです。

data:'The Shawshank Redemption' from 1994

data:'The Godfather' from 1972

data:'The Dark Knight' from 2008

data:'12 Angry Men' from 1957

JVM またはネイティブモードでの実行

開発モードまたはテストモードで実行していない場合は、独自の Kafka ブローカーと Apicurio Registry を起動する必要があります。そのための最も簡単な方法は、docker-compose を使用して適切なコンテナーを起動することです。

Confluent Schema Registry を使用している場合は、すでに Kafka ブローカーと Confluent Schema Registry インスタンスが実行および設定されています。ここに記した`docker-compose` や Apicurio Registry の設定に関する説明は無視してかまいません。

プロジェクトのルートに、次の内容の docker-compose.yaml ファイルを作成します。

version: '2'

services:

  zookeeper:
    image: quay.io/strimzi/kafka:0.39.0-kafka-3.6.1
    command: [
        "sh", "-c",
        "bin/zookeeper-server-start.sh config/zookeeper.properties"
    ]
    ports:
      - "2181:2181"
    environment:
      LOG_DIR: /tmp/logs

  kafka:
    image: quay.io/strimzi/kafka:0.39.0-kafka-3.6.1
    command: [
        "sh", "-c",
        "bin/kafka-server-start.sh config/server.properties --override listeners=$${KAFKA_LISTENERS} --override advertised.listeners=$${KAFKA_ADVERTISED_LISTENERS} --override zookeeper.connect=$${KAFKA_ZOOKEEPER_CONNECT}"
    ]
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
    environment:
      LOG_DIR: "/tmp/logs"
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181

  schema-registry:
    image: apicurio/apicurio-registry-mem:2.4.2.Final
    ports:
      - 8081:8080
    depends_on:
      - kafka
    environment:
      QUARKUS_PROFILE: prod

アプリケーションを開始する前に、まず Kafka ブローカーと Apicurio Registry を起動します。

docker-compose up
コンテナを停止するには、docker-compose down を使用します。docker-compose rm を使用してコンテナをクリーンアップすることもできます。

次のコマンドでアプリケーションをビルドできます。

コマンドラインインタフェース
quarkus build
Maven
./mvnw install
Gradle
./gradlew build

その後、次のコマンドを使用して JVM モードで実行します。

java -Dmp.messaging.connector.smallrye-kafka.apicurio.registry.url=http://localhost:8081/apis/registry/v2 -jar target/quarkus-app/quarkus-run.jar
デフォルトでは、アプリケーションは localhost:9092 でリッスンしている Kafka ブローカーに接続しようとします。ブートストラップサーバーは、java -Dkafka.bootstrap.servers=... -jar target/quarkus-app/quarkus-run.jar を使用して設定できます。

コマンドラインでレジストリー URL を指定する方法は便利ではないため、prod プロファイルに対してのみ設定プロパティーを追加します。

%prod.mp.messaging.connector.smallrye-kafka.apicurio.registry.url=http://localhost:8081/apis/registry/v2

以下でネイティブ実行可能ファイルをビルドできます。

コマンドラインインタフェース
quarkus build --native
Maven
./mvnw install -Dnative
Gradle
./gradlew build -Dquarkus.native.enabled=true

次に、それを実行します。

./target/kafka-avro-schema-quickstart-1.0.0-SNAPSHOT-runner -Dkafka.bootstrap.servers=localhost:9092

アプリケーションのテスト

上記のとおり、Dev Services for Kafka と Apicurio Registry は、Kafka ブローカーと Apicurio Registry インスタンスを開発モードおよびテスト用に自動で開始し、設定します。したがって、Kafka および Apicurio Registry を自分で設定する必要はなく、テストの作成に集中できます。

まず、REST クライアントと Awaitility のテスト依存関係をビルドファイルに追加します。

pom.xml
<!-- we'll use Jakarta REST Client for talking to the SSE endpoint -->
<dependency>
    <groupId>io.quarkus</groupId>
    <artifactId>quarkus-rest-client</artifactId>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>org.awaitility</groupId>
    <artifactId>awaitility</artifactId>
    <scope>test</scope>
</dependency>
build.gradle
testImplementation("io.quarkus:quarkus-rest-client")
testImplementation("org.awaitility:awaitility")

テストでは、ムービーをループで送信し、送信したものを ConsumedMovieResource が返すかどうかを確認します。

package org.acme.kafka;

import io.quarkus.test.common.QuarkusTestResource;
import io.quarkus.test.common.http.TestHTTPResource;
import io.quarkus.test.junit.QuarkusTest;
import io.restassured.http.ContentType;
import org.hamcrest.Matchers;
import org.junit.jupiter.api.Test;

import jakarta.ws.rs.client.Client;
import jakarta.ws.rs.client.ClientBuilder;
import jakarta.ws.rs.client.WebTarget;
import jakarta.ws.rs.sse.SseEventSource;
import java.net.URI;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import static io.restassured.RestAssured.given;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static java.util.concurrent.TimeUnit.SECONDS;
import static org.awaitility.Awaitility.await;
import static org.hamcrest.MatcherAssert.assertThat;

@QuarkusTest
public class MovieResourceTest {

    @TestHTTPResource("/consumed-movies")
    URI consumedMovies;

    @Test
    public void testHelloEndpoint() throws InterruptedException {
        // create a client for `ConsumedMovieResource` and collect the consumed resources in a list
        Client client = ClientBuilder.newClient();
        WebTarget target = client.target(consumedMovies);

        List<String> received = new CopyOnWriteArrayList<>();

        SseEventSource source = SseEventSource.target(target).build();
        source.register(inboundSseEvent -> received.add(inboundSseEvent.readData()));

        // in a separate thread, feed the `MovieResource`
        ExecutorService movieSender = startSendingMovies();

        source.open();

        // check if, after at most 5 seconds, we have at least 2 items collected, and they are what we expect
        await().atMost(5, SECONDS).until(() -> received.size() >= 2);
        assertThat(received, Matchers.hasItems("'The Shawshank Redemption' from 1994",
                "'12 Angry Men' from 1957"));
        source.close();

        // shutdown the executor that is feeding the `MovieResource`
        movieSender.shutdownNow();
        movieSender.awaitTermination(5, SECONDS);
    }

    private ExecutorService startSendingMovies() {
        ExecutorService executorService = Executors.newSingleThreadExecutor();
        executorService.execute(() -> {
            while (true) {
                given()
                        .contentType(ContentType.JSON)
                        .body("{\"title\":\"The Shawshank Redemption\",\"year\":1994}")
                .when()
                        .post("/movies")
                .then()
                        .statusCode(202);

                given()
                        .contentType(ContentType.JSON)
                        .body("{\"title\":\"12 Angry Men\",\"year\":1957}")
                .when()
                        .post("/movies")
                .then()
                        .statusCode(202);

                try {
                    Thread.sleep(200L);
                } catch (InterruptedException e) {
                    break;
                }
            }
        });
        return executorService;
    }

}
プロジェクトと一緒に生成された MovieResourceTest を変更しました。このテストクラスには、ネイティブ実行可能ファイルに対して同じテストを実行するサブクラ ス`NativeMovieResourceIT` があります。実行するには、次を実行します。
コマンドラインインタフェース
quarkus build --native
Maven
./mvnw install -Dnative
Gradle
./gradlew build -Dquarkus.native.enabled=true

手動セットアップ

Dev Services を使用できず、Kafka ブローカーと Apicurio Registry インスタンスを手動で開始する場合は、QuarkusTestResourceLifecycleManager を定義します。

pom.xml
<dependency>
    <groupId>io.strimzi</groupId>
    <artifactId>strimzi-test-container</artifactId>
    <version>0.105.0</version>
    <scope>test</scope>
    <exclusions>
        <exclusion>
            <groupId>org.apache.logging.log4j</groupId>
            <artifactId>log4j-core</artifactId>
        </exclusion>
    </exclusions>
</dependency>
build.gradle
testImplementation("io.strimzi:strimzi-test-container:0.105.0") {
    exclude group: "org.apache.logging.log4j", module: "log4j-core"
}
package org.acme.kafka;

import java.util.HashMap;
import java.util.Map;

import org.testcontainers.containers.GenericContainer;

import io.quarkus.test.common.QuarkusTestResourceLifecycleManager;
import io.strimzi.StrimziKafkaContainer;

public class KafkaAndSchemaRegistryTestResource implements QuarkusTestResourceLifecycleManager {

    private final StrimziKafkaContainer kafka = new StrimziKafkaContainer();

    private GenericContainer<?> registry;

    @Override
    public Map<String, String> start() {
        kafka.start();
        registry = new GenericContainer<>("apicurio/apicurio-registry-mem:2.4.2.Final")
                .withExposedPorts(8080)
                .withEnv("QUARKUS_PROFILE", "prod");
        registry.start();
        Map<String, String> properties = new HashMap<>();
        properties.put("mp.messaging.connector.smallrye-kafka.apicurio.registry.url",
                "http://" + registry.getHost() + ":" + registry.getMappedPort(8080) + "/apis/registry/v2");
        properties.put("kafka.bootstrap.servers", kafka.getBootstrapServers());
        return properties;
    }

    @Override
    public void stop() {
        registry.stop();
        kafka.stop();
    }
}
@QuarkusTest
@QuarkusTestResource(KafkaAndSchemaRegistryTestResource.class)
public class MovieResourceTest {
    ...
}

互換性のあるバージョンのApicurio Registryの使用

quarkus-apicurio-registry-avro エクステンションは、最近のバージョンのApicurio Registryクライアントに依存しており、Apicurio Registryサーバーとクライアントのほとんどのバージョンには後方互換性があります。一部では、Serdesで使用するクライアントがサーバーと互換性があることを確認する必要があります。

例えば、Apicurio dev serviceで、イメージ名をバージョン 2.1.5.Final を使用するように設定した場合:

quarkus.apicurio-registry.devservices.image-name=quay.io/apicurio/apicurio-registry-mem:2.1.5.Final

apicurio-registry-serdes-avro-serde の依存関係と REST クライアント apicurio-common-rest-client-vertx の依存関係が互換性のあるバージョンに設定されていることを確認する必要があります:

pom.xml
<dependency>
    <groupId>io.quarkus</groupId>
    <artifactId>quarkus-apicurio-registry-avro</artifactId>
    <exclusions>
        <exclusion>
            <groupId>io.apicurio</groupId>
            <artifactId>apicurio-common-rest-client-vertx</artifactId>
        </exclusion>
        <exclusion>
            <groupId>io.apicurio</groupId>
            <artifactId>apicurio-registry-serdes-avro-serde</artifactId>
        </exclusion>
    </exclusions>
</dependency>
<dependency>
    <groupId>io.apicurio</groupId>
    <artifactId>apicurio-registry-client</artifactId>
    <version>2.1.5.Final</version>
</dependency>
<dependency>
    <groupId>io.apicurio</groupId>
    <artifactId>apicurio-registry-common</artifactId>
    <version>2.1.5.Final</version>
</dependency>
<dependency>
    <groupId>io.apicurio</groupId>
    <artifactId>apicurio-registry-serdes-avro-serde</artifactId>
    <version>2.1.5.Final</version>
    <exclusions>
        <exclusion>
            <groupId>io.apicurio</groupId>
            <artifactId>apicurio-common-rest-client-jdk</artifactId>
        </exclusion>
        <exclusion>
            <groupId>io.apicurio</groupId>
            <artifactId>apicurio-registry-client</artifactId>
        </exclusion>
        <exclusion>
            <groupId>io.apicurio</groupId>
            <artifactId>apicurio-registry-common</artifactId>
        </exclusion>
    </exclusions>
</dependency>
<dependency>
    <groupId>io.apicurio</groupId>
    <artifactId>apicurio-common-rest-client-vertx</artifactId>
    <version>0.1.5.Final</version>
</dependency>
build.gradle
dependencies {
    implementation(platform("io.quarkus.platform:quarkus-bom:2.12.3.Final"))

    ...

    implementation("io.quarkus:quarkus-apicurio-registry-avro")
    implementation("io.apicurio:apicurio-registry-serdes-avro-serde") {
        exclude group: "io.apicurio", module: "apicurio-common-rest-client-jdk"
        exclude group: "io.apicurio", module: "apicurio-registry-client"
        exclude group: "io.apicurio", module: "apicurio-registry-common"
        version {
            strictly "2.1.5.Final"
        }
    }
    implementation("io.apicurio:apicurio-registry-client") {
        version {
            strictly "2.1.5.Final"
        }
    }
    implementation("io.apicurio:apicurio-registry-common") {
        version {
            strictly "2.1.5.Final"
        }
    }
    implementation("io.apicurio:apicurio-common-rest-client-vertx") {
        version {
            strictly "0.1.5.Final"
        }
    }
}

apicurio-registry-client および apicurio-common-rest-client-vertx の既知の旧互換バージョンは以下の通りです。

  • apicurio-registry-client 2.1.5.Final と 0.1.5.Final apicurio-common-rest-client-vertx

  • apicurio-registry-client 2.3.1.Final と 0.1.13.Final apicurio-common-rest-client-vertx

Confluent Schema Registryの使用

Confluent Schema Registry を使用する場合、quarkus-apicurio-registry-avro エクステンションの代わりに、quarkus-confluent-registry-avro エクステンションと、さらにいくつかの依存関係が必要です。また、Confluent Maven リポジトリーを pom.xml / build.gradle ファイルに追加する必要があります。

pom.xml
<dependencies>
    ...
    <!-- the extension -->
    <dependency>
        <groupId>io.quarkus</groupId>
        <artifactId>quarkus-confluent-registry-avro</artifactId>
    </dependency>
    <!-- Confluent registry libraries use Jakarta REST client -->
    <dependency>
        <groupId>io.quarkus</groupId>
        <artifactId>quarkus-rest-client</artifactId>
    </dependency>
    <dependency>
        <groupId>io.confluent</groupId>
        <artifactId>kafka-avro-serializer</artifactId>
        <version>7.2.0</version>
        <exclusions>
            <exclusion>
                <groupId>jakarta.ws.rs</groupId>
                <artifactId>jakarta.ws.rs-api</artifactId>
            </exclusion>
        </exclusions>
    </dependency>
</dependencies>

<repositories>
    <!-- io.confluent:kafka-avro-serializer is only available from this repository: -->
    <repository>
        <id>confluent</id>
        <url>https://packages.confluent.io/maven/</url>
        <snapshots>
            <enabled>false</enabled>
        </snapshots>
    </repository>
</repositories>
build.gradle
repositories {
    ...

    maven {
        url "https://packages.confluent.io/maven/"
    }
}

dependencies {
    ...

    implementation("io.quarkus:quarkus-confluent-registry-avro")

    // Confluent registry libraries use Jakarta REST client
    implementation("io.quarkus:quarkus-rest-client")

    implementation("io.confluent:kafka-avro-serializer:7.2.0") {
        exclude group: "jakarta.ws.rs", module: "jakarta.ws.rs-api"
    }
}

In JVM mode, any version of io.confluent:kafka-avro-serializer can be used. In native mode, Quarkus supports the following versions: 6.2.x, 7.0.x, 7.1.x, 7.2.x, 7.3.x.

For version 7.4.x and 7.5.x, due to an issue with the Confluent Schema Serializer, you need to add another dependency:

pom.xml
<dependency>
    <groupId>com.fasterxml.jackson.dataformat</groupId>
    <artifactId>jackson-dataformat-csv</artifactId>
</dependency>
build.gradle
dependencies {
    implementation("com.fasterxml.jackson.dataformat:jackson-dataformat-csv")
}

それ以外のバージョンでは、ネイティブコンフィギュレーションの調整が必要な場合があります。

Avro コード生成の詳細

このガイドでは、Quarkus コード生成メカニズムを使用して、Avro スキーマから Java ファイルを生成しました。

内部で、メカニズムは org.apache.avro:avro-compiler を使用します。

次の設定プロパティーを使用して、動作を変更できます。

  • avro.codegen.[avsc|avdl|avpr].imports - 最初にコンパイルされるべきファイルやディレクトリのリストで、後にコンパイルされるスキーマからインポートできるようにします。インポートされたファイルは互いに参照しあってはならないことに注意してください。すべてのパスは src/[main|test]/avro ディレクトリ、またはビルドシステムによって設定された任意のソースディレクトリの avro サブディレクトリからの相対パスでなければなりません。カンマで区切られたリストとして渡されます。

  • avro.codegen.stringType - Avro 文字列に使用する Java タイプ。CharSequenceStringUtf8 のいずれかである可能性があります。デフォルトは String です。

  • avro.codegen.createOptionalGetters - 要求されたタイプの Optional を返す`getOptional…​` メソッドを生成できるようにします。デフォルトは false です。

  • avro.codegen.enableDecimalLogicalType - 10 進型に Java クラスを使用するかどうかを決定します。デフォルトは false です。

  • avro.codegen.createSetters - レコードのフィールドにセッターを作成するかどうかを決定します。デフォルトは false です。

  • avro.codegen.gettersReturnOptional - 要求されたタイプの Optional を返す get…​ メソッドを生成できるようにします。デフォルトは false です。

  • avro.codegen.optionalGettersForNullableFieldsOnly は、gettersReturnOptional オプションと連携して機能します。設定されている場合、Optional ゲッターは null 許容フィールドに対してのみ生成されます。フィールドが必須の場合、通常のゲッターが生成されます。デフォルトは false です。

さらに詳しく

関連コンテンツ