スキーマのリンク¶
Schema Linking は、2 つの Schema Registry クラスター間でスキーマの同期を継続する Confluent の機能です。
Schema Linking を Cluster Linking と併用すると、2 つの Schema Registry および Kafka クラスターの間でスキーマとトピックデータの両方を同期し続けることができます。
Schema Linking を Cluster Linking と併用せず単独で使用して、スキーマの集約、バックアップ、ステージング、移行を目的としてクラスター間でスキーマを複製することもできます。
Schema Linking を使用するには スキーマエクスポーター が必要です。スキーマエクスポーターは Schema Registry にあり、同一の Schema Registry クラスター内または異なる Schema Registry クラスターで、あるコンテキストから別のコンテキストへと継続的にスキーマをエクスポートします。
スキーマエクスポーターは、スキーマコンテキスト と呼ばれるグループでスキーマを同期できます。スキーマコンテキストとは、スキーマ ID とサブジェクト名を個別にグループ化したものです。スキーマがコンテキストなしでエクスポートされる場合（contextType: NONE）、これらのスキーマはそのままエクスポートされ、デフォルトのコンテキストに入ります。
Schema Linking 機能の詳細については、「Schema Linking」を参照してください。
Schema Linking を実行するワークフローの概要を以下に示します。
送信元の Schema Registry クラスターでスキーマを定義します。
送信元クラスターでスキーマを登録するときに、スキーマ名にコンテキストを挿入することでカスタムコンテキストを指定できます。コンテキストが指定されていない場合、デフォルトのコンテキストが使用されます。
送信元の Schema Registry クラスターにスキーマエクスポーターを作成します。
エクスポートされたスキーマは、送信先の Schema Registry に
IMPORTモードで配置されます。IMPORTモードのスキーマに変更を加えることはできません。必要に応じて以下の操作を行います。
Confluent for Kubernetes （CFK）には、スキーマエクスポーターの作成と管理のワークフロー全体をサポートする宣言型 API、SchemaExporter のカスタムリソース定義（CRD）が用意されています。
Schema Registry でのスキーマエクスポーターの有効化¶
送信元の Schema Registry の CR をアップデートしてスキーマエクスポーターを有効にしてから、kubectl apply -f <Schema Registry CR> コマンドで変更を適用します。
spec:
passwordEncoder: --- [1]
enableSchemaExporter: --- [2]
- [1]（省略可）送信元の Schema Registry のパスワードエンコーダーを指定します。詳細については、「パスワードエンコーダーシークレットの管理」を参照してください。
- [2] Schema Registry でスキーマエクスポーターを有効にするには
trueに設定します。
スキーマエクスポーターの作成¶
スキーマエクスポーターの作成と管理は送信元の Schema Registry クラスターで行います。
送信元の Schema Registry クラスターでスキーマエクスポーターの CR を作成し、kubectl apply -f <Schema Exporter CR> コマンドで構成を適用します。
apiVersion: platform.confluent.io/v1beta1
kind: SchemaExporter
metadata:
name: --- [1]
namespace: --- [2]
spec:
sourceCluster: --- [3]
destinationCluster: --- [4]
subjects: --- [5]
subjectRenameFormat: --- [6]
contextType: --- [7]
contextName: --- [8]
configs: --- [9]
[1]（必須）スキーマエクスポーターの名前。送信元の Schema Registry クラスターで一意の名前である必要があります。
[2] スキーマエクスポーターの名前空間。
[3] 送信元の Schema Registry クラスター。クラスター名またはエンドポイントを指定できます。指定されていない場合、CFK はこのスキーマエクスポーターの名前空間で送信元の Schema Registry を自動検出します。名前空間で複数の Schema Registry クラスターが検出された場合、エラーが発生して検出プロセスが中断されます。
構成の詳細については、「送信元および送信先の Schema Registry クラスターの指定」を参照してください。
[4] スキーマがエクスポートされる送信先の Schema Registry クラスター。定義されていない場合、送信元クラスターが送信先として使用され、スキーマエクスポーターは送信元クラスター内のコンテキスト全体にスキーマをエクスポートします。
構成の詳細については、「送信元および送信先の Schema Registry クラスターの指定」を参照してください。
[5] 送信先にエクスポートするサブジェクト。デフォルト値は、デフォルトのコンテキストのすべてのサブジェクトを示す
["*"]です。[6] 送信先のサブジェクトの名前を変更する方法を定義する名前変更形式。
たとえば、値が
my-${subject}の場合、送信先のサブジェクトはmy-XXXとなります（XXXは元のサブジェクトを示します）。[7] サブジェクトを送信先に移動するためのコンテキストの作成方法を指定します。
デフォルト値は
AUTOです。この場合、エクスポーターは送信先クラスターで自動生成されたコンテキストを使用します。自動生成されたコンテキストの名前はステータスで報告されます。NONEに設定した場合、エクスポーターは送信元スキーマをそのままコピーします。[8] サブジェクトをエクスポートする送信先のスキーマコンテキストの名前。この名前が定義されている場合、
spec.contextTypeは無視されます。[9] SchemaExporter の CRD のプロパティでは追加の構成はサポートされていません。
SchemaExporter の CR の例:
apiVersion: platform.confluent.io/v1beta1
kind: SchemaExporter
metadata:
name: schemaExporter
namespace: operator
spec:
sourceCluster:
schemaRegistryClusterRef:
name: sr
namespace: operator
destinationCluster:
schemaRegistryRest
endpoint: https://schemaregistry.operator-dest.svc.cluster.local:8081
authentication:
type: basic
secretRef: sr-basic
subjects:
- subject1
- subject2
contextType: CUSTOM
contextName: link-source
送信元および送信先の Schema Registry クラスターの指定¶
スキーマエクスポーターは、以下のいずれかの方法によって、送信元と送信先の Schema Registry クラスターを指定できます。
Schema Registry クラスター名を使用した Schema Registry の指定¶
スキーマエクスポーターに送信元または送信先の Schema Registry を指定するには、SchemaExporter の CR の spec.sourceCluster または spec.destinationCluster に以下を設定します。
schemaRegistryClusterRef:
name: --- [1]
namespace: --- [2]
- [1]（必須）|sr| クラスターの名前。
- [2]（省略可）スキーマの名前空間と異なる場合、Schema Registry クラスターが実行されている名前空間を指定します。
Schema Registry エンドポイントを使用した Schema Registry の指定¶
Schema Registry エンドポイントへの接続方法を指定するには、SchemaExporter の CR の spec.sourceCluster または spec.destinationCluster セクションに接続情報を指定します。
Schema Registry エンドポイント
schemaRegistryRest:
endpoint: --- [1]
authentication:
type: --- [2]
- [1] Schema Registry が実行されているエンドポイント。
- [2] Schema Registry クラスターに使用する認証方式。サポートされるタイプは、
basic、mtls、およびbearerです。Schema Registry で RBAC が有効になっている場合にbearerを使用できます。
Schema Registry に対する基本認証
schemaRegistryRest:
authentication:
type: basic --- [1]
basic:
secretRef: --- [2]
directoryPathInContainer: --- [3]
[1] 基本認証タイプの場合に必須です。
[2]（必須） 認証情報を格納するシークレットの名前。必要な形式については、「基本認証」を参照してください。
[3] Vault によって必要な認証情報が挿入されるコンテナーのディレクトリパスに設定します。
必要な形式については、「基本認証」を参照してください。
Vault を使用する際の認証情報と必要なアノテーションを提供する方法については、「Confluent Platform アプリケーションの CR へのシークレットの提供」を参照してください。
Schema Registry に対する mTLS 認証
schemaRegistryRest:
authentication:
type: mtls --- [1]
tls:
secretRef: --- [2]
directoryPathInContainer: --- [3]
[1] mTLS 認証タイプの場合に必須です。
[2] TLS 証明書を格納するシークレットの名前。
TLS シークレットの想定されるキーについては、「Confluent for Kubernetes でのネットワーク暗号化の構成」を参照してください。
[3] Vault によって TLS 証明書が挿入されるコンテナーのディレクトリパスに設定します。
想定されるキーと証明書については、「Confluent for Kubernetes でのネットワーク暗号化の構成」を参照してください。
Vault を使用する際の認証情報と必要なアノテーションを提供する方法については、「Confluent Platform アプリケーションの CR へのシークレットの提供」を参照してください。
Schema Registry に対するベアラー認証（RBAC の場合）
Schema Registry で RBAC が有効になっている場合、ベアラー認証を次のように構成できます。
schemaRegistryRest:
authentication:
type: bearer --- [1]
bearer:
secretRef: --- [2]
directoryPathInContainer: --- [3]
[1] ベアラー認証タイプの場合に必須です。
[2] または [3] が必須です。
[2] ベアラー認証情報を格納するシークレットの名前。必要な形式については、「ベアラー認証」を参照してください。
[3] Vault によって必要な認証情報が挿入されるコンテナーのディレクトリパスに設定します。
必要な形式については、「ベアラー認証」を参照してください。
Vault を使用する際の認証情報と必要なアノテーションを提供する方法については、「Confluent Platform アプリケーションの CR へのシークレットの提供」を参照してください。
Schema Registry クラスターの TLS 暗号化
tls:
enabled: true --- [1]
secretRef: --- [2]
directoryPathInContainer: --- [3]
[1]（必須）
[2] または [3] が必須です。
[2] 証明書を格納するシークレットの名前。証明書の詳細については、「Confluent for Kubernetes でのネットワーク暗号化の構成」を参照してください。
[3] Vault によって TLS 証明書が挿入されるコンテナーのディレクトリパスに設定します。
想定されるキーと証明書については、「Confluent for Kubernetes でのネットワーク暗号化の構成」を参照してください。
Vault を使用する際の認証情報と必要なアノテーションを提供する方法については、「Confluent Platform アプリケーションの CR へのシークレットの提供」を参照してください。
スキーマエクスポーターの構成の編集¶
既存のエクスポーターの構成をアップデートすると、CFK がエクスポーターを一時停止し、構成をアップデートして、エクスポーターを再開します。
既存のエクスポーターについては、構成の以下のプロパティを変更できません。既存のエクスポーターを削除してエクスポーターを再作成する必要があります。
- 送信元の Schema Registry
- 送信先の Schema Registry
- スキーマエクスポーターの名前
スキーマエクスポーターの CR を必要な構成になるように編集し、kubectl apply -f <Schema Exporter CR> コマンドで適用します。
コンテキストタイプ（contextType）のデフォルトは、作成中のみ AUTO です。カスタムコンテキストでスキーマエクスポーターを作成し、自動生成されたコンテキストを使用するように編集する場合は、contextType を明示的に AUTO に設定する必要があります。
コンテキスト名（contextName）が編集されると、新しいサブジェクト / スキーマのみが新しいコンテキストにエクスポートされます。アップデート前に同期された古いスキーマは前のコンテキストで同期されます。古いスキーマすべてを新しいコンテキストに移行するには、エクスポーターをリセットする必要があります。
同様に、subjectRename 形式が編集された場合、新しいスキーマのみが新しい名前の形式で移行されます。エクスポーターをリセットし、既に同期されているスキーマを新しい名前の形式で再移行する必要があります。
スキーマエクスポーターのリセット¶
スキーマエクスポーターの状態は、STARTING、RUNNING、PAUSED のいずれかです。
スキーマエクスポーターをリセットすると、保存済みのオフセットがクリアされます。
スキーマエクスポーターをリセットするには、以下のコマンドでエクスポーターのリセットのアノテーションをスキーマエクスポーターの CR に追加します。
kubectl annotate schemaexporter schemaexporter platform.confluent.io/reset-schema-exporter="true"
スキーマエクスポーターの一時停止¶
スキーマエクスポーターを一時停止するには、以下のコマンドでエクスポーターの一時停止のアノテーションをスキーマエクスポーターの CR に追加します。
kubectl annotate schemaexporter schemaexporter platform.confluent.io/pause-schema-exporter="true".
スキーマエクスポーターの再開¶
スキーマエクスポーターを再開するには、以下のコマンドでエクスポーターの再開のアノテーションをスキーマエクスポーターの CR に追加します。
kubectl annotate schemaexporter schemaexporter platform.confluent.io/resume-schema-exporter="true".
スキーマエクスポーターの削除¶
スキーマエクスポーターを削除しても、既に送信先にエクスポートされているスキーマは削除されません。送信先の Schema Registry にエクスポートされたスキーマは、同期された最新の状態にとどまります。
スキーマリンクが切断された場合、「スキーマの移行」で説明されている移行を使用して、エクスポートされたスキーマの IMPORT モードを解除できます。
スキーマの IMPORT モードを解除したら、送信先の Schema Registry でこれらのスキーマを管理できるように、送信先クラスターでこれらのスキーマ用の CR を作成します。
スキーマエクスポーターを削除するには、次のコマンドを実行します。
kubectl delete schemaexporter schemaexporter.