> ## Documentation Index
> Fetch the complete documentation index at: https://docs.streamnative.io/llms.txt
> Use this file to discover all available pages before exploring further.

# Validate Schema IDs on the Broker

> Configure the broker to reject records whose schema ID isn't registered for the topic's subject, so bad data never reaches the log.

Client-side serializers register schemas, but nothing stops an application from bypassing them and
producing arbitrary bytes to a schematized topic. Broker-side schema ID validation closes that gap:
the broker inspects each record's schema ID prefix and rejects the record if that ID doesn't resolve
to a schema registered under the topic's expected subject.

<Note title="What it checks">
  Validation confirms that the record carries a schema ID and that the ID maps to a registered schema
  under the subject the naming strategy derives. It does **not** inspect the payload or verify that the
  data matches the schema.
</Note>

## Enable validation on a topic

Validation is off by default and configured per topic through two properties:

| Property                            | Default | Effect                                    |
| ----------------------------------- | ------- | ----------------------------------------- |
| `kop.kafka.key.schema.validation`   | `false` | Validate the schema ID on message keys.   |
| `kop.kafka.value.schema.validation` | `false` | Validate the schema ID on message values. |

Set them when you create the topic.

<Tabs>
  <Tab title="snctl">
    ```bash theme={null}
    snctl kafka admin topics create my-topic-sv \
      --partitions 4 \
      --config kop.kafka.value.schema.validation=true
    ```
  </Tab>

  <Tab title="pulsar-admin">
    ```bash theme={null}
    bin/pulsar-admin \
        --admin-url <Pulsar Service HTTP URL> \
        --auth-plugin org.apache.pulsar.client.impl.auth.AuthenticationToken \
        --auth-params token:<API Key> \
      topics create-partitioned-topic persistent://public/default/my-topic-sv \
        -p 4 \
        -m kop.kafka.value.schema.validation=true
    ```
  </Tab>
</Tabs>

With validation on, a message whose value carries no schema ID—or carries one that doesn't match a
schema registered for the subject—is rejected. The broker returns an error to the producer and
discards the message.

## Choose a subject name strategy

The broker derives the subject to validate against from a naming strategy. The default is
`TopicNameStrategy`. Change it with:

| Property                                | Applies to     |
| --------------------------------------- | -------------- |
| `kop.kafka.key.subject.name.strategy`   | Message keys   |
| `kop.kafka.value.subject.name.strategy` | Message values |

| Strategy                  | Subject derived from                                                                                          |
| ------------------------- | ------------------------------------------------------------------------------------------------------------- |
| `TopicNameStrategy`       | The topic name. For topic `my-topic`, the subject is `my-topic-value` for values and `my-topic-key` for keys. |
| `RecordNameStrategy`      | The fully qualified record name.                                                                              |
| `TopicRecordNameStrategy` | The topic name plus the fully qualified record name.                                                          |

Pass the fully qualified Confluent class name:

<Tabs>
  <Tab title="snctl">
    ```bash theme={null}
    snctl kafka admin topics create my-other-topic-sv \
      --partitions 4 \
      --config kop.kafka.value.schema.validation=true \
      --config kop.kafka.value.subject.name.strategy=io.confluent.kafka.serializers.subject.RecordNameStrategy
    ```
  </Tab>

  <Tab title="pulsar-admin">
    ```bash theme={null}
    bin/pulsar-admin \
        --admin-url <Pulsar Service HTTP URL> \
        --auth-plugin org.apache.pulsar.client.impl.auth.AuthenticationToken \
        --auth-params token:<API Key> \
      topics create-partitioned-topic persistent://public/default/my-other-topic-sv \
        -p 4 \
        -m kop.kafka.value.schema.validation=true \
        -m kop.kafka.value.subject.name.strategy=io.confluent.kafka.serializers.subject.RecordNameStrategy
    ```
  </Tab>
</Tabs>

Set the same strategy on your producers, so client and broker derive the same subject.

<Warning title="RecordNameStrategy and subject parsing">
  Subject names are parsed into Pulsar tenant, namespace, and topic components, so a record name like
  `com.acme.MyRecord` becomes tenant `com`, namespace `acme`, topic `MyRecord`. Before using
  `RecordNameStrategy` or `TopicRecordNameStrategy`, read
  [Subject names map to Pulsar coordinates](/kafka/governance/sr/reference/confluent-compatibility#subject-names-map-to-pulsar-coordinates).
</Warning>

## Limitations

<Note title="Tombstones always pass">
  Validation doesn't reject tombstone records—messages with a null value—even when no schema ID is
  attached. This is deliberate: it keeps deletes working on compacted topics instead of being blocked
  by validation.
</Note>

These properties use the `kop.kafka.*` prefix rather than Confluent's `confluent.key.schema.validation`
and `confluent.value.schema.validation`. Configuration copied from Confluent won't take effect.

## Next steps

<CardGroup cols={2}>
  <Card title="Serializers and deserializers" icon="code" href="/kafka/governance/sr/fundamentals/serdes">
    Subject naming strategies and the record wire format.
  </Card>

  <Card title="Confluent compatibility" icon="list-check" href="/kafka/governance/sr/reference/confluent-compatibility">
    Behavior differences to check before you migrate.
  </Card>
</CardGroup>