Contents
GCPでスキーマレジストリを作成する際の注意点
GCPではgcloud CLIやコンソールでスキーマレジストリを管理できますが、特にalphaリリースは安定性に注意が必要です。一部環境では利用不可となるため、betaまたはstableバージョンのコマンドを使用することを強く推奨します。
GCPコンソールでの初期設定
- Google Cloud Consoleにログインし、「Managed Service for Apache Kafka」サービスを選択します。
- 「スキーマレジストリ」メニューから「新規作成」をクリックします。
- スキーマ名、バージョン、互換性ポリシー（例:
BACKWARD）を指定し、リソースを作成します。
注意: ここでのポリシー設定は後述の「BACKWARDポリシーの適用」と密接に関連するため、慎重に選択してください。
gcloud CLIによる自動化（安定性重視版）
以下のようにCLIで一括作成が可能です。スクリプト化やCI/CDでの利用を想定しています。alphaリリースではなくbetaまたはstableバージョンを使用することを前提とします。
|
1 2 3 4 5 6 |
gcloud beta managed-kafka schemas create \ --cluster=your-cluster-name \ --schema-id=my-schema \ --schema-type=AVRO \ --compatibility=BACKWARD |
| 項目 | 値 | 補足 |
|---|---|---|
| クラスター | your-cluster-name | 事前に作成済のクラスター名 |
| スキーマID | my-schema | ユニークな識別子 |
| 形式 | AVRO | JSONやPROTOなど選択可能 |
| 互換性ポリシー | BACKWARD | システム全体の設計に影響 |
重要:
alphaリリースは一部環境で利用不可となるため、安定性を求める場合はbetaやstableバージョンを使用してください。
Apicurio RegistryとRed Hat Streamsの統合方法
Red Hat Streams for Apache Kafkaでは、Apicurio Registryを活用してスキーマの一元管理を行うことができます。この統合はKafkaプロダーコンシューマの設定とポリシー設計に影響を与えるため、慎重な構成が求められます。
Kafkaプロダーコンシューマ設定
Apicurio Registryを有効化するには、schema.registry.urlをApicurioのエンドポイントに指定します。以下はJavaアプリケーションにおける例です。
|
1 2 3 |
Properties props = new Properties(); props.put("schema.registry.url", "http://apicurio-registry:8080"); |
互換性ポリシーの適用
ApicurioではBACKWARDやFULLなどのポリシーを選択できますが、Red Hat Streamsのドキュメントによると、Kafka StreamsはBACKWARDのみをサポートします。これは、メッセージのフィールド追加時のデフォルト値指定を強制するためです。
重要:
BACKWARDポリシーでは、新規フィールドにデフォルト値を明示的に設定しないと、旧バージョンのコンシューマがエラーを発生させる可能性があります。たとえば、オプションフィールドであればnullを指定します。
設定ファイル例（Apicurio + Red Hat Streams）
以下は、Apicurio Registryとの統合に必要な設定ファイルの例です。この設定により、スキーマ検証と互換性ポリシーが自動的に適用されます。
|
1 2 3 4 5 6 7 8 |
apicurio: registry: url: "http://apicurio-registry:8080" compatibility: "BACKWARD" kafka: streams: application-id: "my-stream-app" |
AWS Lambdaでのスキーマ検証設定
AWS LambdaはKafkaメッセージをリアルタイム処理する際、スキーマレジストリを活用してエラーを事前に検出できます。これにより、変更されたスキーマによる破損が防止されます。
Lambdaトリガーの構成手順（具体的な例）
- AWS Management ConsoleからLambda関数を作成し、「Kafkaイベントソース」を選択します。
- スキーマレジストリのURLを指定して、事前に定義されたスキーマに基づく検証を有効化します。
Lambdaトリガー設定例（CLIコマンド）
|
1 2 3 4 5 |
aws lambda create-event-source-mapping \ --function-name my-lambda-function \ --event-source-arn arn:aws:kafka:region:account-id:topic/my-topic \ --schema-register-url http://schema-registry.example.com |
エラーハンドリングのベストプラクティス
- スキーマ不一致時のロギング: 以下のコードで例外をキャッチし、CloudWatch Logsに記録します。
|
1 2 3 4 5 |
try: # スキーマ検証処理 except SchemaValidationError as e: print(f"Schema validation failed: {e}") |
- レトロフィット対応: 旧バージョンのスキーマをサポートする場合は、
BACKWARDポリシーを採用し、デフォルト値を明示します。
BACKWARDポリシーの適用とデフォルト値指定
BACKWARDポリシーは、新規スキーマが旧バージョンとの互換性を持つことを保証する仕組みです。ただし、設計ミスにより運用に深刻な影響を及ぼす可能性もあるため、慎重な設定が必要です。
ポリシー選択の根拠
- デフォルト値必須の理由: Kafka Streamsは新規フィールドの追加を許容しますが、既存の消費者（旧バージョン）にはそのフィールドが存在しない場合があります。このため、デフォルト値を明示的に指定することで、デシリアライズエラーを防止できます。
互換性チェックロジックの設計
- フィールド追加: デフォルト値を定義（例:
default: null） - フィールド削除や型変更: ポリシーでは許可されないため、システム全体への影響評価が不可欠です。
運用時のリスク回避策
| シナリオ | 対応策 |
|---|---|
| 旧バージョンのコンシューマ | BACKWARDポリシーで安全なデシリアライズを保証 |
| スキーマ変更時 | CI/CDに自動検証ステップを組み込む |
Confluent Schema Registryのバージョン管理機能
Confluent独自のスキーマレジストリでは、バージョン管理が高度に実装されています。これにより、過去のスキーマ履歴を可視化し、自動的な更新も可能です。
スキーマ履歴の可視化（UI手順）
- Confluent Schema RegistryのUIにログインします。
- 「Subjects」メニューから対象のトピックを選択します。
- 「Versions」タブをクリックし、各バージョンのスキーマ定義や変更履歴を確認できます。
自動的なバージョンアップデート（手順）
- 新規スキーマを登録
- システム全体に通知される（
GET /subjects/my-topic/versionsAPIで確認可） - ポリシーに基づいて自動承認または却下（UIから手動レビューも可能）
注意: 自動更新は、ポリシー設定の妥当性を事前に確認済みであることが前提です。運用ミスを防ぐためにも、手動レビューを併せて行いましょう。
まとめ: 実環境での検証が不可欠な理由
本記事では、スキーマレジストリの設定プロセスとベストプラクティスをクラウドとオンプレミス共通の観点から解説しました。特に以下のポイントに注意してください:
- Google CloudやAWSそれぞれでCLIの自動化が有効
- Apicurio Registryとの統合時は
BACKWARDポリシーとデフォルト値設定を徹底 - AWS Lambdaでのリアルタイム検証は、エラーハンドリング設計に直結
- Confluentのバージョン管理機能は、運用効率を高めるが慎重な操作が必要
各クラウドサービスの公式ドキュメントを参照しつつ、本記事の設定手順を実環境で検証してください。