Kafkaのトピックとスキーマのバインディング設計: 互換性違反を防ぐ実務

Kafkaブローカー自体はメッセージの構造を検証しません。実運用では、Schema Registryと命名戦略、互換性レベルを組み合わせて「どのトピックにどのスキーマが来るべきか」を強制する仕組み作りが重要です。

本稿では、トピックとスキーマの結び付け（バインディング）を中心に、互換性違反を未然に防ぐための設計と運用を解説します。Confluent CCDAKで頻出のSubject命名戦略、互換性レベル、ブローカー側検証の要点も押さえます。

なぜトピックとスキーマを結び付けるのか

Kafkaは高スループットのログであり、ブローカーはメッセージ構造を知りません。スキーマ検証を担うのはクライアントのシリアライザ/デシリアライザとSchema Registryです。従って、トピックとスキーマの関係を曖昧にすると、破壊的変更がすり抜け、本番でデシリアライズ失敗や集計ジョブの停止を招きます。

バインディング設計の核は、Subject命名戦略で“どの単位でスキーマ互換性を管理するか”を決め、Registry側で互換性を強制し、必要に応じてブローカー側検証を有効化することです。さらに、CI/CDで事前に互換性チェックを行い、レジストリ登録時点で破壊的変更を拒否できるようにします。

• Kafkaブローカーは構造検証をしない。検証はSerializer/Registry/（オプションで）ブローカー側検証が担う
• Subject命名戦略が“互換性の束ね方”を決める
• 互換性違反を防ぐには、登録前チェック、Registry互換性、ブローカー側検証の三段構えが有効

エンドツーエンドの検証ポイント

Subject命名戦略の選び方

Schema Registryでは、登録単位であるSubjectの命名戦略によって互換性の束ね方が変わります。典型はTopicNameStrategy（デフォルト）で、<topic>-value に値スキーマをひも付けます。複数のイベント型を1トピックに混在させる場合は、RecordNameStrategyやTopicRecordNameStrategyが実務で有効です。

CCDAKでは、どの戦略がどのユースケースに適するか、混在時の落とし穴、同じトピックに異なるイベント型を入れるときの選択肢が問われがちです。

• 単一イベント型/トピック: TopicNameStrategyが分かりやすく運用も容易
• 複数イベント型/トピック: TopicRecordNameStrategyで型ごとに互換性を独立管理
• レコード名を安定運用するため、Avroのnamespace/名前を組織内で標準化する

クライアントでのSubject命名戦略指定（Java例）

Properties p = new Properties();
p.put("key.serializer", "io.confluent.kafka.serializers.KafkaAvroSerializer");
p.put("value.serializer", "io.confluent.kafka.serializers.KafkaAvroSerializer");
p.put("schema.registry.url", "https://registry:8081");
// トピック内に複数イベント型を入れる場合
p.put("value.subject.name.strategy",
      "io.confluent.kafka.serializers.subject.TopicRecordNameStrategy");

互換性レベルと進化ルール

Schema Registryはグローバルとサブジェクト単位の互換性レベルを持ちます。一般的な既定はBACKWARD（後方互換）で、最新スキーマで旧データを読むことを保証します。運用では多くの場合、BACKWARD_TRANSITIVE（過去すべてのバージョンとの互換）を推奨します。

Avro/JSON Schema/Protobufは進化ルールに差異がありますが、実務では「フィールドの追加はデフォルト値を付ける」「必須フィールドの削除・型変更は避ける」が共通の安全策です。トピック内に複数型がある場合でも、TopicRecordNameStrategyなら型ごとに独立して互換性チェックできます。

• 後方互換で安全: フィールド追加（デフォルト値あり）、ユニオンへ拡張（互換性に注意）
• 破壊的になりがち: 必須フィールド削除、型の縮小（long→int）、フィールド意味の変更
• TRANSITIVEは“全バージョン”に対する互換性を要求。検出漏れを防ぐので本番向き
• グローバル設定は初期値。最終的にはサブジェクト単位で明示設定する

互換性の設定と事前チェック（REST API）

# グローバル互換性をBACKWARD_TRANSITIVEに
curl -s -X PUT -H 'Content-Type: application/json' \
  --data '{"compatibility":"BACKWARD_TRANSITIVE"}' \
  https://registry:8081/config

# サブジェクト単位で上書き
curl -s -X PUT -H 'Content-Type: application/json' \
  --data '{"compatibility":"FULL_TRANSITIVE"}' \
  https://registry:8081/config/orders-value

# 登録前に互換性をテスト（v=latestと新スキーマの組合せを検証）
curl -s -X POST -H 'Content-Type: application/json' \
  --data '{"schemaType":"AVRO","schema":"{\\"type\\":\\"record\\",...}"}' \
  https://registry:8081/compatibility/subjects/orders-value/versions/latest

互換性違反を防ぐCI/CDと環境分離

最も多い事故は“レビューをすり抜けた破壊的変更のレジストリ登録”です。回避策は、スキーマをアプリとは独立したアーティファクトとして管理し、PR時にRegistryへの互換性チェックを行い、失敗したらマージを止めることです。登録権限はCIのロボットユーザーに限定します。

環境分離はレジストリを環境ごとに分け、DEV→STG→PRDへスキーマを昇格させます。IDは環境間で一致しない前提で、内容（フィンガープリント）で同一性を判断します。

• スキーマを単独リポジトリで管理。タグ付けとリリースノートで進化を可視化
• PRで /compatibility を呼び出し、破壊的なら失敗
• CIだけが /subjects へのPOSTを実行。人手登録は禁止
• 本番レジストリは読み取りを広く、書き込みを厳格に。監査ログを必ず残す

CIジョブ例（擬似）: 互換性チェック→合格なら登録

# 1) 互換性テスト
curl -f -X POST -H 'Content-Type: application/json' \
  --data @candidate-schema.json \
  https://registry:8081/compatibility/subjects/orders-value/versions/latest

# 2) 合格後に登録
curl -f -X POST -H 'Content-Type: application/json' \
  --data @candidate-schema.json \
  https://registry:8081/subjects/orders-value/versions

実行時の強制: ブローカー検証、Connect、ksqlDB

Confluent Serverでは、トピック単位でブローカー側スキーマ検証を有効化できます。これにより、Schema Registryに存在しないスキーマIDや互換性違反のメッセージをブローカーで拒否できます。クリティカルなトピックでは有効化を検討してください。

Kafka Connectでは、Avro/JSON Schema/Protobufの各ConverterをSchema Registry連携で設定し、schemas.enableを有効にします。ksqlDBは対象トピックのスキーマに依存するため、進化時は下流のストリーム/テーブルに影響が及ぶかを事前に評価します。

• ブローカー側検証: confluent.key.schema.validation / confluent.value.schema.validation をトピックで有効化
• Connect: AvroConverter/ProtobufConverter/JsonSchemaConverter + schema.registry.url を設定
• ksqlDB: 永続クエリの互換性を確認。破壊的変更は事前にマイグレーション手順を用意

トピックでブローカー側スキーマ検証を有効化（Confluent Server）

# 既存トピックに付与
kafka-configs --bootstrap-server broker:9092 \
  --alter --topic orders \
  --add-config confluent.value.schema.validation=true,confluent.key.schema.validation=true

# Connectの例（Avro）
"key.converter":"io.confluent.connect.avro.AvroConverter",
"value.converter":"io.confluent.connect.avro.AvroConverter",
"key.converter.schema.registry.url":"https://registry:8081",
"value.converter.schema.registry.url":"https://registry:8081",
"value.converter.schemas.enable": true

監視とインシデント対応の型

互換性違反は登録時点で止めるのが最善ですが、実行時の失敗に備えた観測性も必要です。典型はコンシューマのデシリアライズ例外、ブローカー拒否、Registryの409エラーです。DLQとメトリクス、監査ログを組み合わせて早期検知と切り戻しを可能にします。

切り戻しは“スキーマを戻す”か“コンシューマを新スキーマ対応に更新する”かの二択です。前者は過去バージョンを最新として再登録、後者は段階的ロールアウト（カナリア）で影響を抑えます。

• Registryの4xx/5xx率、スキーマ登録件数、互換性チェック失敗件数を可視化
• コンシューマのデシリアライズ例外をDLQへ退避し、件数アラート
• スキーマのロールバック手順書を用意。過去バージョンの再アクティベーションを訓練

問題で確認

よくある質問