スキーマ検証 on Confluent Server¶

Kafka のトピックに対して生成されたデータに、サブジェクト命名方法に従ってスキーマレジストリに登録された有効なスキーマ ID が使用されていることは、ブローカーがスキーマ検証を通じて確認します。

ちなみに

スキーマ検証 is also available on Confluent Cloud clusters using the hosted, per environment スキーマレジストリ. To learn more, see Using Broker-Side Schema Validation on Confluent Cloud.

前提条件とスキーマレジストリの URL の設定¶

以下の例を実行するための基本的な要件は、 Schema Registry のチュートリアルに記載されている条件とほぼ同じです。ただし、ここでは Maven は必要ありません。また、Confluent Platform バージョン 5.4.0 以降が必要となります。

ブローカーのスキーマ検証を有効にするためのもう 1 つの前提条件として、Confluent Platform を起動する前に、Kafka の server.properties ファイル（$CONFLUENT_HOME/etc/kafka/server.properties）で confluent.schema.registry.url を指定する必要があります。

以下に例を示します。

confluent.schema.registry.url=http://schema-registry:8081

この構成には、スキーマレジストリインスタンスの URL をコンマ区切りのリストで指定することができます。Confluent CLI と Confluent Control Center の両方からスキーマ検証を利用するために、この設定が必要となります。

Confluent CLI からトピックのスキーマ検証を有効にする¶

トピックのスキーマ検証は、トピックの作成時または既存のトピックの変更時に有効にすることができます。

スキーマ検証を使用してトピックを作成する¶

トピックの作成時にそのスキーマ検証を設定するには、confluent.value.schema.validation=true と confluent.key.schema.validation=true を設定します。

値スキーマの検証とキースキーマの検証は互いに独立しています。一方のみを有効にすることも、両方を有効にすることもできます。（デフォルトでは、スキーマ検証は有効になっていません。キースキーマと値スキーマの検証はどちらもデフォルトで false になっています。）

たとえば、次のコマンドでは my-topic-sv というトピックが作成され、値スキーマに対するスキーマ検証が有効になります。

kafka-topics --create --bootstrap-server localhost:9092 --replication-factor 1 \
--partitions 1 --topic my-topic-sv \
--config confluent.value.schema.validation=true

このコマンドの出力は次のとおりです。

Created topic my-topic-sv.

この構成で、メッセージの値の有効なスキーマを持たないメッセージがトピック my-topic-sv に生成された場合、プロデューサーにエラーが返されて、メッセージが破棄されます。

バッチで送信されたメッセージの中に無効なメッセージが 1 つでもあった場合、そのバッチ全体が破棄されます。

ちなみに

値スキーマとキースキーマの詳細については、スキーマレジストリのチュートリアルの「用語の復習」を参照してください。

既存のトピックにスキーマ検証を追加する¶

my-first-topic という新しいトピックを作成します。

kafka-topics --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic my-first-topic

このコマンドの出力は次のとおりです。

Created topic my-first-topic.

既存のトピックに対する検証構成を（このケースでは、false から true に）変更するには、alter フラグと --add-config フラグを使用して検証を指定します。その例を次に示します。

kafka-configs --bootstrap-server localhost:9092 --alter --entity-type topics --entity-name my-first-topic --add-config confluent.value.schema.validation=true

次のような確認メッセージが表示されます。

Completed updating config for topic 'my-first-topic'.

このトピックに対する検証を無効（true から false）にするには、上記のコマンドに --add-config confluent.value.schema.validation=false を指定して再度実行します。

トピックのサブジェクト命名方法を変更する¶

スキーマレジストリ内のスキーマにトピックをマップするための命名方法として、デフォルトでは TopicNameStrategy が Confluent Server によって使用されます。

Confluent Platform 5.5.0 未満では、サブジェクト命名方法がブローカーの server.properties で構成されており、ブローカー上のすべてのトピックに同じ命名方法を使用する必要がありました。

Confluent Platform 5.5.0 以降では、命名方法がトピックに関連付けられます。そのため、スキーマのサブジェクトのキーと値の両方に、デフォルトの "トピックごと" 以外の命名方法を選択できるようになりました。その構成には、confluent.key.subject.name.strategy と confluent.value.subject.name.strategy を使用します。

Confluent CLI から、--config オプションに命名方法を指定してトピックを作成または変更します。以下に例を示します。

RecordNameStrategy を値に使用するトピックを作成するには、次のようにします。

./bin/kafka-topics --create --bootstrap-server localhost:9092 \
--replication-factor 1 --partitions 1 --topic my-other-cool-topic \
--config confluent.value.schema.validation=true --config confluent.value.subject.name.strategy=io.confluent.kafka.serializers.subject.RecordNameStrategy

RecordNameStrategy をキーに使用するようにトピックを変更するには、次のようにします。

kafka-configs --bootstrap-server localhost:9092 \
--alter --entity-type topics --entity-name my-other-cool-topic \
--add-config confluent.key.subject.name.strategy=io.confluent.kafka.serializers.subject.RecordNameStrategy

次の構成は、confluent.value.subject.name.strategy と confluent.key.subject.name.strategy のどちらにでも使用できます。

io.confluent.kafka.serializers.subject.TopicNameStrategy （デフォルト）
io.confluent.kafka.serializers.subject.RecordNameStrategy
io.confluent.kafka.serializers.subject.TopicRecordNameStrategy

参考

スキーマ、サブジェクト、トピック

Confluent Control Center でトピックのスキーマ検証を有効にする¶

トピックのスキーマ検証は、Control Center で表示したり、その有効と無効を切り替えたりすることができます。加えて、トピックごとのサブジェクト命名方法を構成することができます。

トピックのスキーマ検証設定を表示または変更する¶

Control Center （ http://localhost:9021/ ）から現在の構成を表示したりトピックのスキーマ検証を有効にしたりするには、次の手順に従います。

既存のトピックの Configuration タブをクリックします。
Switch to expert mode をクリックします。
エキスパートモードで、confluent.value.schema.validation と confluent.key.schema.validation の設定を false から true に変更します。
Save changes をクリックします。

トピックのサブジェクト命名方法を変更する¶

Control Center を使用してサブジェクト命名方法を変更するには、次の手順に従います。

Control Center で、アップデートするトピックを選択し、Configuration、Switch to expert mode を順にクリックします。
設定を変更し、Save changes をクリックします。

次の構成は、confluent.value.subject.name.strategy と confluent.key.subject.name.strategy のどちらにでも使用できます。

io.confluent.kafka.serializers.subject.TopicNameStrategy （デフォルト）
io.confluent.kafka.serializers.subject.RecordNameStrategy
io.confluent.kafka.serializers.subject.TopicRecordNameStrategy

詳細については、「トピックのサブジェクト命名方法を変更する」を参照してください。同じタスクを Confluent CLI で行う方法について説明しているほか、この機能（Confluent Platform 5.5.0 の新機能）の背景情報についても詳しく説明しています。

デモ : コマンドラインからトピックのスキーマ検証を有効にする¶

この短いデモでは、トピックのスキーマ検証を有効または無効にした場合の効果を紹介しています。

Confluent Platform とスキーマレジストリを使い始めて間もない方は、最初に「スキーマレジストリのチュートリアル」を読んでから、このデモをご覧になることをお勧めします。

以下の例では、$CONFLUENT_HOME/bin にある kafka-console-producer と kafka-console-consumer を利用しています。

Confluent Platform バージョン 5.4.0 以降のローカルインストールでは、$CONFLUENT_HOME/etc/kafka/server.properties に、次のスキーマレジストリ URL の構成を追加します。
```
############################## My Schema Validation Demo Settings ################
# Schema Registry URL
confluent.schema.registry.url=http://localhost:8081
```
上の例では、ファイルの構成を追跡するために 2 行のコメントが含まれていますが、これらは省略してもかまいません。
次のコマンドを使用して Confluent Platform を起動します。
```
confluent local services start
```
ちなみに
- 単に confluent local schema-registry を実行してもかまいません。その場合は、依存関係として kafka と zookeeper も起動されます。このデモでは、Connect や Control Center など、他のサービスを直接参照することはしていません。ただし、さらに理解を深めるために（トピックやメッセージが Control Center にどのように表示されるかなど）、フルスタックを実行する場合もあるでしょう。confluent local の詳細については、「Confluent Platform を使用した Apache Kafka のクイックスタート（ローカル）」および Confluent CLI コマンドリファレンスの「confluent local」を参照してください。
- confluent local のコマンドはバックグラウンドで実行されるので、そのコマンドウィンドウを再利用することができます。プロデューサーとコンシューマーのセッションは独立している必要があります。
test-schemas というテストトピックを作成します。デフォルトの false に設定されるようスキーマ検証設定は指定しません。
```
kafka-topics --bootstrap-server localhost:9092 --create --partitions 1 --replication-factor 1 --topic test-schemas
```
これでトピックが作成されます。テストトピックに生成されたレコードに対するブローカー検証は行われません。このデモの最初のパートでは、意図的にそのようにしてあります。トピックが作成されたことは、kafka-topics --bootstrap-server localhost:9092 --list を使用して確認できます。
プロデューサーの新しいコマンドウィンドウで次のコマンドを実行し、（デフォルトの文字列シリアライザーを使用して）シリアル化されたレコードをトピック test-schemas に生成します。
```
kafka-console-producer --broker-list localhost:9092 --topic test-schemas --property parse.key=true --property key.separator=,
```
現在このトピックはスキーマ検証が無効になっているためこのコマンドは成功します。先ほどこのトピックに対してブローカーのスキーマ検証を有効にしていた場合、そのトピックを生成先とする上のコマンドは許可されません。

このコマンドを実行すると、プロデューサーのコマンドプロンプト（> ）が出力として表示されます。ここで、生成するメッセージを入力することができます。

> プロンプトに、初めてのメッセージを次のように入力します。
```
1,my first record
```
このプロデューサーセッションは実行したままにしてください。
コンシューマー用に新しいコマンドウィンドウを開き、次のコマンドを入力してメッセージを読み取ります。
```
kafka-console-consumer --bootstrap-server localhost:9092 --from-beginning --topic test-schemas --property print.key=true
```
このコマンドからは my first record が出力されます。

このコンシューマーセッションは実行したままにしてください。
今度は、トピック test-schemas のスキーマ検証を true に設定します。
```
kafka-configs --bootstrap-server localhost:9092 --alter --entity-type topics --entity-name test-schemas --add-config confluent.value.schema.validation=true
```
Completed updating config for topic test-schemas. という確認のメッセージが表示されます。
プロデューサーセッションに戻り、> プロンプトに 2 つ目のメッセージを入力します。
```
2,my second record
```
スキーマ検証が有効になっており送信しているメッセージにはスキーマ ID が含まれていないので、This record has failed the validation on broker というエラーが返されます。

If you subsequently disable スキーマ検証 (use the same command to set it to false), restart the producer, then type and resend the same or another similarly formatted message, the message will go through. (For example, produce 3,my third record.)

ちなみに

As an alternative to restarting the producer in this last step (after switching schema validation back to false), you can simply type or copy-paste the suggested message 3,my third record on the blank line after the error and hit return. The consumer will pick it up, and you will then get the producer prompt back. However, this is an unintuitive workflow because there is no prompt showing after the error, and you may be tempted to hit return first, which will shut down the producer.

正常に生成されたメッセージは、Control Center （ウェブブラウザーで http://localhost:9021/ にアクセス）で、Topics、test-schemas、messages の順に選択して表示される画面にも表示されます。以前に送信されたメッセージを表示するには、パーティションを選択するか、タイムスタンプに移動しなければならない場合があります。
シャットダウンとクリーンアップのタスクを実行します。
- コンシューマーとプロデューサーを停止するには、それぞれのコマンドウィンドウで Ctl キーを押しながら C キーを押します。
- Confluent Platform を停止するには、「 confluent local stop」と入力します。
- 別のテストを最初から実行する前に既存のデータ（トピック、スキーマ、メッセージ）を削除する場合は、「 confluent local destroy 」と入力します。

Configuring Security for スキーマ検証¶

In general, スキーマレジストリ initiates the connection to the brokers. スキーマ検証 is unique in that the broker(s) initiate the connection to スキーマレジストリ. They do so in order to retrieve schemas from the registry, and verify that the messages they receive from producers match schemas associated with particular topics. With スキーマ検証 enabled, the sequence of tasks looks something like this:

A broker receives a message from a producer, and sees that it's directed to a topic that has a schema associated.
The broker initiates a connection to スキーマレジストリ.
The broker asks for the schema associated with the topic (by schema ID).
スキーマレジストリ receives the request, finds the requested schema in its schema storage, and returns it to the broker.
The broker validates the message against the schema.

Therefore, to set up security on a cluster that has broker-side スキーマ検証 enabled on topics, you must configure settings on the Kafka broker to support this broker-initiated connection to スキーマレジストリ. For multiple brokers, each broker must be configured. For example, for mTLS, ideally you would have a different certificate for each broker.

Note that Schema Registry's internal Kafka client to Kafka brokers is not relevant at all to the connection between broker-side スキーマ検証 and Schema Registry's HTTP listeners. The security settings below do not reflect anything about the スキーマレジストリ internal client-to-broker connection.

The broker configurations below include confluent.schema.registry.url, which tells the broker how to connect to スキーマレジストリ. You may already have configured this on your brokers, as a prerequisite for using Schema Validation. The rest of the settings shown are specific to security configurations.

ちなみに

The sections below focus on security configurations on the brokers. For information about setting up security on スキーマレジストリ, see スキーマレジストリのセキュリティの概要 and related sections. For example, to set up スキーマレジストリ to require clients (like brokers) to authenticate with a username and password, see スキーマレジストリ under HTTP 基本認証.

mTLS (mutual TLS) Authentication¶

To configure two-way authentication with SSL (also known as mTLS), the broker will authenticate to スキーマレジストリ using SSL for authentication (certificates) and encryption.

Define the following settings in the broker properties file for each broker ($CONFLUENT_HOME/etc/kafka/server.properties).

confluent.schema.registry.url=https://<host>:<port>
confluent.ssl.truststore.location=<path to truststore file>
confluent.ssl.truststore.password=<password for the truststore if encrypted>
confluent.ssl.keystore.location=<path to file with private key for the broker>
confluent.ssl.keystore.password=<password to keystore file>
confluent.ssl.key.password=<password>

Basic Authentication¶

For this setup, the brokers are configured to authenticate to スキーマレジストリ using basic authentication.

Define the following settings on each broker ($CONFLUENT_HOME/etc/kafka/server.properties).

confluent.schema.registry.url=http://<host>:<port>
confluent.basic.auth.credentials.source=<USER_INFO, URL, or SASL_INHERIT>
confluent.basic.auth.user.info=<username>:<password> #required only if credentials source is set to USER_INFO

The property confluent.basic.auth.credentials.source defines the type of credentials to use (user name and passsword). These are literals, not variables.
If you set confluent.basic.auth.credentials to USER_INFO, you must also specify confluent.basic.auth.user.info.

Basic Authentication with SSL¶

Define the following settings on each broker ($CONFLUENT_HOME/etc/kafka/server.properties).

confluent.schema.registry.url=https://<host>:<port>
confluent.basic.auth.credentials.source=USER_INFO
confluent.basic.auth.user.info=<username>:<password>
confluent.ssl.truststore.location=<path to the truststore file>
confluent.ssl.truststore.password=<password for the truststore if encrypted>

Role-based Access Control (RBAC) and ACLs¶

You can configure スキーマレジストリのロールベースアクセス制御の構成 (and/or ACLs) for authorization to スキーマレジストリ, and use it with other types of security on the brokers.

The スキーマレジストリセキュリティプラグイン supports authorization for both RBAC and ACLs.

Configure the Brokers¶

Define the appropriate settings on the brokers similar to those shown in previous sections, depending on the type of security. For example, for basic authentication, the broker properties files would contain:

confluent.schema.registry.url=http://<host>:<port>
confluent.basic.auth.credentials.source=USER_INFO
confluent.basic.auth.user.info=<username>:<password> #required only if credentials source is set to USER_INFO

Set RBAC Role Bindings to Allow Access to スキーマレジストリ¶

At a minimum, in terms of defining RBAC role bindings, スキーマレジストリ needs the following IAM assignment:

confluent iam rolebinding list --principal User:<user-id> \
--role DeveloperRead --kafka-cluster-id <kafka-cluster-id> \
--resource Subject:* --schema-registry-cluster-id <schema-registry-group-id>

The スキーマレジストリ cluster ID is the same as schema-registry-group-id, which defaults to schema-registry.

Suggested Resources¶

ブログの投稿記事 : Confluent Platform 5.4 でのスキーマ検証
スキーマ検証 and Replicator explains how スキーマ検証 interacts with Replicator configurations used for replicating data from one cluster to another

スキーマ検証 on Confluent Server¶

前提条件と スキーマレジストリ の URL の設定¶

Confluent CLI からトピックの スキーマ検証 を有効にする¶

スキーマ検証 を使用してトピックを作成する¶

既存のトピックに スキーマ検証 を追加する¶

トピックのサブジェクト命名方法を変更する¶

Confluent Control Center でトピックの スキーマ検証 を有効にする¶

トピックの スキーマ検証 設定を表示または変更する¶

トピックのサブジェクト命名方法を変更する¶

デモ : コマンドラインからトピックの スキーマ検証 を有効にする¶

Configuring Security for スキーマ検証¶

mTLS (mutual TLS) Authentication¶

Basic Authentication¶

Basic Authentication with SSL¶

Role-based Access Control (RBAC) and ACLs¶

Configure the Brokers¶

Set RBAC Role Bindings to Allow Access to スキーマレジストリ¶

Suggested Resources¶

前提条件とスキーマレジストリの URL の設定¶

Confluent CLI からトピックのスキーマ検証を有効にする¶

スキーマ検証を使用してトピックを作成する¶

既存のトピックにスキーマ検証を追加する¶

Confluent Control Center でトピックのスキーマ検証を有効にする¶

トピックのスキーマ検証設定を表示または変更する¶

デモ : コマンドラインからトピックのスキーマ検証を有効にする¶