Schema RegistryとAvro/Protobufを使ったKafka + Spring Boot

1. 概要

複数のサービスが同じKafkaトピックに対して読み書きする場合、最大のリスクはメッセージフォーマットです。単にJSONフィールド名をproducerで変更するだけで、後段のconsumerが破損する可能性があります。Confluent Schema Registryはこの問題を解決します。スキーマを一元管理し、各メッセージにスキーマIDを付け、互換性を損なう変更をブロックします。

このガイドでは、Spring Boot 3.xでSchema Registryをセットアップし、AvroとProtobufを使ってメッセージを送受信する方法、互換性設定、実際のKafkaなしでテストを書く方法を説明します。

2. Schema Registryの仕組み

Schema RegistryはKafkaクラスタの隣に独立したRESTサービスとして動作します。内部ではスキーマを_schemasトピックに保存します。各スキーマはsubjectに属します。デフォルト（TopicNameStrategy）では、ordersトピック内のvalueのsubjectはorders-valueという名前になります。

producerがメッセージを送信する時、serializerは次の処理を行います：

1. Registry上でスキーマを登録またはルックアップしてスキーマIDを取得します。結果はキャッシュされます。
2. Confluentのwire formatでペイロードを書き込みます：

| byte 0: magic byte (0) | byte 1-4: schema ID (int) | byte 5..: Avro/Protobufバイナリデータ |

consumerは最初の5バイトを読んでスキーマIDを取得し、スキーマを1回ロードしてからキャッシュに入れ、その後ペイロードをデコードします。メッセージはスキーマ全体を含む必要がないため、JSONと比べてサイズが大幅に小さくなります。

3. ローカル環境をセットアップ

Docker ComposeでKafka（KRaftモード）とSchema Registryを実行します：

services:
  kafka:
    image: confluentinc/cp-kafka:7.7.1
    ports: ["9092:9092"]
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:29093
      KAFKA_LISTENERS: PLAINTEXT://kafka:29092,CONTROLLER://kafka:29093,EXTERNAL://0.0.0.0:9092
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,EXTERNAL://localhost:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      CLUSTER_ID: MkU3OEVBNTcwNTJENDM2Qk

  schema-registry:
    image: confluentinc/cp-schema-registry:7.7.1
    depends_on: [kafka]
    ports: ["8081:8081"]
    environment:
      SCHEMA_REGISTRY_HOST_NAME: schema-registry
      SCHEMA_REGISTRY_KAFKA_STORE_BOOTSTRAP_SERVERS: kafka:29092
      SCHEMA_REGISTRY_LISTENERS: http://0.0.0.0:8081

docker compose up -dを実行した後、Registryを確認できます：

curl -s http://localhost:8081/subjects

[]

結果は空のリストです。まだsubjectが登録されていないためです。

4. Spring BootでAvro

4.1. 依存関係とプラグイン

Confluent系serializerはMaven Centralにはありません。Confluentのリポジトリを追加する必要があります：

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

<dependencies>
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
    </dependency>
    <dependency>
        <groupId>io.confluent</groupId>
        <artifactId>kafka-avro-serializer</artifactId>
        <version>7.7.1</version>
    </dependency>
    <dependency>
        <groupId>org.apache.avro</groupId>
        <artifactId>avro</artifactId>
        <version>1.12.0</version>
    </dependency>
</dependencies>

avro-maven-pluginは.avscファイルからJavaクラスを生成します：

<plugin>
    <groupId>org.apache.avro</groupId>
    <artifactId>avro-maven-plugin</artifactId>
    <version>1.12.0</version>
    <executions>
        <execution>
            <phase>generate-sources</phase>
            <goals><goal>schema</goal></goals>
            <configuration>
                <sourceDirectory>${project.basedir}/src/main/avro</sourceDirectory>
                <outputDirectory>${project.build.directory}/generated-sources/avro</outputDirectory>
                <stringType>String</stringType>
            </configuration>
        </execution>
    </executions>
</plugin>

stringType=Stringオプションにより、文字列フィールドはCharSequenceではなくjava.lang.Stringとして生成されます。このおかげで、コードやテスト内での比較がシンプルになります。

4.2. スキーマ定義

src/main/avro/Order.avscを作成します：

{
  "type": "record",
  "name": "Order",
  "namespace": "dev.nonla.order.avro",
  "fields": [
    { "name": "id", "type": "string" },
    { "name": "customerId", "type": "string" },
    { "name": "amount", "type": "double" }
  ]
}

mvn generate-sourcesを実行します。プラグインはdev.nonla.order.avro.Orderクラスを生成します。このクラスにはbuilderとスキーマが組み込まれています。

4.3. producerとconsumerの設定

spring:
  kafka:
    bootstrap-servers: localhost:9092
    properties:
      schema.registry.url: http://localhost:8081
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: io.confluent.kafka.serializers.KafkaAvroSerializer
    consumer:
      group-id: shipping-service
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: io.confluent.kafka.serializers.KafkaAvroDeserializer
      properties:
        specific.avro.reader: true

schema.registry.urlプロパティはspring.kafka.properties以下に配置するため、producerとconsumerの両方に適用されます。specific.avro.reader: trueプロパティは非常に重要です。これがないと、consumerはOrderクラスではなくGenericRecordを受け取ります。

4.4. メッセージの送受信

@Service
public class OrderPublisher {

    private final KafkaTemplate<String, Order> kafkaTemplate;

    public OrderPublisher(KafkaTemplate<String, Order> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public void publish(String orderId, String customerId, double amount) {
        Order order = Order.newBuilder()
                .setId(orderId)
                .setCustomerId(customerId)
                .setAmount(amount)
                .build();
        kafkaTemplate.send("orders", order.getId(), order);
    }
}

@Component
public class ShippingListener {

    private static final Logger log = LoggerFactory.getLogger(ShippingListener.class);

    @KafkaListener(topics = "orders")
    public void onOrder(Order order) {
        log.info("注文{}を顧客{}に配送準備中、金額: {}",
                order.getId(), order.getCustomerId(), order.getAmount());
    }
}

初回送信時、KafkaAvroSerializerは自動的にスキーマをsubject orders-valueに登録します。確認できます：

curl -s http://localhost:8081/subjects/orders-value/versions/latest

{"subject":"orders-value","version":1,"id":1,"schema":"{\"type\":\"record\",\"name\":\"Order\",...}"}

この時点で、トピック上のメッセージはスキーマID 1のみを含みます。スキーマ全体は含まれません。

5. Spring BootでProtobuf

Protobufの手順はAvroとほぼ同じで、serializerとコード生成方法が異なるだけです。以下の依存関係を追加します：

<dependency>
    <groupId>io.confluent</groupId>
    <artifactId>kafka-protobuf-serializer</artifactId>
    <version>7.7.1</version>
</dependency>

src/main/proto/order_event.protoを定義します：

syntax = "proto3";

package dev.nonla.order;

option java_package = "dev.nonla.order.proto";
option java_multiple_files = true;

message OrderEvent {
  string id = 1;
  string customer_id = 2;
  double amount = 3;
}

JavaクラスはMaven用のprotocプラグイン（例：io.github.ascopes:protobuf-maven-plugin）を使用して生成します。その後、serializerを設定します：

spring:
  kafka:
    producer:
      value-serializer: io.confluent.kafka.serializers.protobuf.KafkaProtobufSerializer
    consumer:
      value-deserializer: io.confluent.kafka.serializers.protobuf.KafkaProtobufDeserializer
      properties:
        specific.protobuf.value.type: dev.nonla.order.proto.OrderEvent

specific.protobuf.value.typeプロパティはdeserializerに対象クラスを指定します。このプロパティがないと、DynamicMessageを受け取ります。

どちらのフォーマットを選ぶべき？

• AvroはKafkaエコシステムとデータパイプライン（Kafka ConnectやksqlDB）に適しています。スキーマはJSON形式で、進化ルールはフィールド名とデフォルト値に基づいています。
• Protobufは既にgRPCを使用している組織に適しています。既存の.protoファイルを再利用できます。進化ルールはフィールドの通し番号に基づいています。

6. スキーマの進化と互換性

6.1. 互換性レベル

各subjectには互換性レベルがあります。デフォルトはBACKWARDです。

レベル	意味	デプロイの安全な順序
BACKWARD	新スキーマが古いデータを読める	consumerが先、producerが後
FORWARD	古いスキーマが新しいデータを読める	producerが先、consumerが後
FULL	両方向互換	どちらの順序でもOK
*_TRANSITIVE	直前のバージョンだけでなく、すべての古いバージョンとの互換性をチェック	同様

各subjectのレベルはREST APIで設定します：

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

6.2. 有効な変更の例

Orderにcurrencyフィールドを追加したいとします。Avroでは、新フィールドはデフォルト値を持つ必要があると、新スキーマがFULL互換性を満たします：

{ "name": "currency", "type": "string", "default": "VND" }

コードをマージする前に、新スキーマをRegistryと検証する必要があります：

curl -s -X POST -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  --data '{"schema": "{\"type\":\"record\",\"name\":\"Order\",\"namespace\":\"dev.nonla.order.avro\",\"fields\":[{\"name\":\"id\",\"type\":\"string\"},{\"name\":\"customerId\",\"type\":\"string\"},{\"name\":\"amount\",\"type\":\"double\"},{\"name\":\"currency\",\"type\":\"string\",\"default\":\"VND\"}]}"}' \
  http://localhost:8081/compatibility/subjects/orders-value/versions/latest

{"is_compatible":true}

"default": "VND"を省くと、結果はfalseになります。その場合、producerはスキーマ登録時に409 Conflictエラーを受けます。

Protobufでは、対応するルールは：フィールド番号を変更または再利用しないことです。フィールドを削除する場合は、reservedでマーク付けして誰も番号を再利用できないようにします：

message OrderEvent {
  reserved 2;
  reserved "customer_id";
  string id = 1;
  double amount = 3;
  string currency = 4;
}

7. mock Schema Registryでテストを書く

Confluentのserializerはmock://<scope>形式のURLをサポートしています。このURLを使うと、serializerはメモリ内の疑似Registryを使用します。実際のDockerなしでserialize処理をテストできます：

import static org.assertj.core.api.Assertions.assertThat;

class OrderAvroSerdeTest {

    private static final Map<String, Object> CONFIG = Map.of(
            "schema.registry.url", "mock://order-serde-test",
            "specific.avro.reader", true);

    @Test
    void shouldSerializeWithWireFormatAndDeserializeBack() {
        try (var serializer = new KafkaAvroSerializer();
             var deserializer = new KafkaAvroDeserializer()) {
            serializer.configure(CONFIG, false);
            deserializer.configure(CONFIG, false);

            Order order = Order.newBuilder()
                    .setId("ORD-1001")
                    .setCustomerId("CUS-42")
                    .setAmount(250_000.0)
                    .build();

            byte[] bytes = serializer.serialize("orders", order);

            assertThat(bytes[0]).isZero();                       // マジックバイト
            assertThat(ByteBuffer.wrap(bytes, 1, 4).getInt()).isPositive(); // スキーマID

            Object result = deserializer.deserialize("orders", bytes);
            assertThat(result).isInstanceOf(Order.class).isEqualTo(order);
        }
    }
}

このテストは3つのことを確認しています：

• 最初のバイトはmagic byte 0です。
• 次の4バイトは有効なスキーマIDです。
• deserializerはGenericRecordではなく、正しくOrderクラスを返します。これでspecific.avro.readerが機能していることが確認できます。

より完全な統合テストの場合、TestcontainersでKafkaとSchema Registryを実行できます。しかし、mock://を使ったテストは速く、ほとんどの設定エラーを捕捉するに十分です。

注意： 同じscope mock://を使うserializerは同じ疑似Registryを共有します。テストクラスごとに固有のscopeを使って、相互の影響を避けましょう。

8. よくあるエラーと非推奨ユースケース

Poison pill. wire format正しくないメッセージは、KafkaAvroDeserializerがlistenerに到達する前にSerializationExceptionをスローします。consumerは永遠にそのoffsetで止まる可能性があります。ErrorHandlingDeserializerでdeserializerをラップすることをお勧めします：

spring:
  kafka:
    consumer:
      value-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
      properties:
        spring.deserializer.value.delegate.class: io.confluent.kafka.serializers.KafkaAvroDeserializer

Production環境でのスキーマ自動登録。 デフォルトではauto.register.schemas=trueなので、どのproducerでも新しいスキーマを登録できます。Production環境ではauto.register.schemas: falseとuse.latest.version: trueを設定してください。スキーマはCI/CDパイプラインを通じてのみ登録される必要があり、互換性チェック後に行うべきです。

意図せずGenericRecordまたはDynamicMessageを受け取る。 通常はspecific.avro.readerまたはspecific.protobuf.value.typeを忘れています。結果として、listener内でClassCastExceptionが発生します。

Avroのnamespaceまたはレコード名を変更。 Avroではこれは完全に異なるスキーマとみなされます。Specific readerは対応するクラスを見つけられません。namespaceはコントラクトの変わらない部分と考えてください。

互換性をNONEに設定して楽をする。 これはRegistryのすべてのメリットを失うことになります。試験用トピックにのみNONEを使用してください。

注意： Schema Registryはランタイム依存関係です。serializerはキャッシュを使用するため、Registryが一時的に停止していても、既知のスキーマにはほぼ影響ありません。しかし新しいproducerが起動したり、キャッシュにないスキーマがあったりすると、すぐにエラーになります。Production環境では、Registryを複数インスタンスで実行してください。

Schema Registryを使用すべきでない場合：

• システムが小規模で、1つのチームだけがproducerとconsumerの両方を所有しており、変更が少ない。この場合、SpringのJsonSerializerでJSONを使う方がずっと簡単です。
• メッセージがConfluent wire formatを理解しないクライアント（例えばサードパーティツール）に直接読まれる必要がある場合は、純粋なJSONのみです。
• インフラストラクチャサービスを追加で運用できず、マネージドRegistryもない場合。

9. まとめ

Schema RegistryはKafkaメッセージフォーマットを、バージョン管理と自動検証を備えたコントラクトに変えます。ここまで、以下の方法を見てきました：

• Dockerを使ってRegistryをセットアップ。
• Spring Boot 3でKafkaAvroSerializerとKafkaProtobufSerializerを使ってメッセージを送受信。
• magic byteとスキーマIDから成るwire formatを理解。
• 互換性レベルを設定し、安全にスキーマを進化：Avroではデフォルト付き新フィールド、Protobufではreserved。
• mock:// Registryで素早くテストを実行。

Production環境に展開する際は、スキーマの自動登録を無効にし、互換性チェックをCIに組み込み、常にdeserializerをErrorHandlingDeserializerでラップしてください。そうすることで、スキーマ変更エラーはレビュー時に発見され、深夜にconsumerが破損することはありません。