> ## 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.

# Manage Pulsar Schemas

> Upload, retrieve, and delete Pulsar schemas, and configure the compatibility strategy, AutoUpdate, and schema validation enforcement.

Manage schemas in the Pulsar Schema Registry with `pulsarctl`, the Pulsar admin REST API, or the
`PulsarAdmin` Java client.

## Prerequisites

* A Pulsar cluster on StreamNative Cloud.
* An API key or OAuth credentials for a service account with admin access to the target tenant and
  namespace. See [Service accounts](/cloud/security/authentication/service-accounts/service-accounts).
* [`pulsarctl`](/tools/cli/pulsarctl/pulsarctl-overview) installed and configured, if you use the CLI.

<Note title="Note">
  The examples below use `pulsarctl`. The equivalent `pulsar-admin` commands take the same subcommands
  and arguments—pass `--admin-url`, `--auth-plugin org.apache.pulsar.client.impl.auth.AuthenticationToken`,
  and `--auth-params token:<API Key>` to point it at your cluster.
</Note>

## Upload a schema

To register a new schema for a topic:

<Tabs>
  <Tab title="pulsarctl">
    ```bash theme={null}
    pulsarctl schemas upload --filename <schema-definition-file> <topic-name>
    ```
  </Tab>

  <Tab title="REST API">
    Send a `POST` request to the schema endpoint. This example posts a payload stored in `schema.json`:

    ```bash theme={null}
    curl -X POST \
      -H 'Content-Type: application/json' \
      -H "Authorization: Bearer $TOKEN" \
      -d @schema.json \
      https://<pulsar-cluster-http-url>/admin/v2/schemas/my-tenant/my-ns/my-topic/schema
    ```
  </Tab>

  <Tab title="Java">
    ```java theme={null}
    PulsarAdmin admin = // ...

    PostSchemaPayload payload = new PostSchemaPayload();
    payload.setType("INT8");
    payload.setSchema("");

    admin.createSchema("my-tenant/my-ns/my-topic", payload);
    ```
  </Tab>
</Tabs>

The schema definition file and the REST payload use the same JSON format:

```json theme={null}
{
    "type": "<schema-type>",
    "schema": "<an-utf8-encoded-string-of-schema-definition-data>",
    "properties": {}
}
```

| Field        | Description                                                                                                                                                                                                                                                                      |
| ------------ | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `type`       | For primitive schemas, one of the values in [Primitive types](/cloud/governance/sr/pulsar/schema-types#primitive-types). For struct schemas: `AVRO`, `PROTOBUF`, `PROTOBUF_NATIVE`, or `JSON`.                                                                                   |
| `schema`     | The schema definition data, UTF-8 encoded. For `AVRO`, `PROTOBUF`, and `JSON`, this is an [Avro schema definition](https://avro.apache.org/docs/1.11.1/specification/) in JSON format. For `PROTOBUF_NATIVE`, it's a Protobuf descriptor. For primitive schemas, leave it blank. |
| `properties` | Additional properties associated with the schema.                                                                                                                                                                                                                                |

Here's a complete JSON schema example:

```json theme={null}
{
    "type": "JSON",
    "schema": "{\"type\":\"record\",\"name\":\"User\",\"namespace\":\"com.foo\",\"fields\":[{\"name\":\"file1\",\"type\":[\"null\",\"string\"],\"default\":null},{\"name\":\"file2\",\"type\":[\"null\",\"string\"],\"default\":null},{\"name\":\"file3\",\"type\":[\"string\",\"null\"],\"default\":\"dfdf\"}]}",
    "properties": {}
}
```

## Get a schema

### Get the latest version

<Tabs>
  <Tab title="pulsarctl">
    ```bash theme={null}
    pulsarctl schemas get <topic-name>
    ```

    Example output:

    ```json theme={null}
    {
        "version": 0,
        "type": "String",
        "timestamp": 0,
        "data": "string",
        "properties": {
            "property1": "string",
            "property2": "string"
        }
    }
    ```
  </Tab>

  <Tab title="REST API">
    ```bash theme={null}
    curl -H "Authorization: Bearer $TOKEN" \
      https://<pulsar-cluster-http-url>/admin/v2/schemas/my-tenant/my-ns/my-topic/schema
    ```
  </Tab>

  <Tab title="Java">
    ```java theme={null}
    PulsarAdmin admin = // ...

    SchemaInfo si = admin.getSchema("my-tenant/my-ns/my-topic");
    ```
  </Tab>
</Tabs>

### Get a specific version

<Tabs>
  <Tab title="pulsarctl">
    ```bash theme={null}
    pulsarctl schemas get <topic-name> --version <version>
    ```
  </Tab>

  <Tab title="REST API">
    ```bash theme={null}
    curl -H "Authorization: Bearer $TOKEN" \
      https://<pulsar-cluster-http-url>/admin/v2/schemas/my-tenant/my-ns/my-topic/schema/1
    ```
  </Tab>

  <Tab title="Java">
    ```java theme={null}
    PulsarAdmin admin = // ...

    SchemaInfo si = admin.getSchema("my-tenant/my-ns/my-topic", 1L);
    ```
  </Tab>
</Tabs>

## Delete a schema

<Warning>
  Deleting a schema removes **all versions** registered for the topic. There's no per-version delete
  and no way to recover the versions afterward.
</Warning>

<Tabs>
  <Tab title="pulsarctl">
    ```bash theme={null}
    pulsarctl schemas delete <topic-name>
    ```
  </Tab>

  <Tab title="REST API">
    ```bash theme={null}
    curl -X DELETE -H "Authorization: Bearer $TOKEN" \
      https://<pulsar-cluster-http-url>/admin/v2/schemas/my-tenant/my-ns/my-topic/schema
    ```

    The response returns the version that was current at deletion:

    ```json theme={null}
    {
        "version": "<the-latest-version-number-of-the-schema>"
    }
    ```
  </Tab>

  <Tab title="Java">
    ```java theme={null}
    PulsarAdmin admin = // ...

    admin.deleteSchema("my-tenant/my-ns/my-topic");
    ```
  </Tab>
</Tabs>

## Set the compatibility strategy

The [compatibility strategy](/cloud/governance/sr/pulsar/compatibility#compatibility-strategies) can
be configured at the topic and namespace levels, and **the topic level wins**. If you set it at both,
the topic-level value applies.

<Note title="Cluster-level configuration">
  Pulsar also supports a cluster-level default through broker configuration. On StreamNative Cloud that
  default is managed for you—set the strategy at the topic or namespace level instead.
</Note>

### Topic level

<Tabs>
  <Tab title="pulsarctl">
    ```bash theme={null}
    pulsarctl topicPolicies set-schema-compatibility-strategy <strategy> <topic-name>
    ```
  </Tab>

  <Tab title="Java">
    ```java theme={null}
    PulsarAdmin admin = // ...

    admin.topicPolicies().setSchemaCompatibilityStrategy(
        "my-tenant/my-ns/my-topic", SchemaCompatibilityStrategy.ALWAYS_INCOMPATIBLE);
    ```
  </Tab>
</Tabs>

### Namespace level

<Tabs>
  <Tab title="pulsarctl">
    ```bash theme={null}
    pulsarctl namespaces set-schema-compatibility-strategy --compatibility <strategy> <tenant>/<namespace>
    ```
  </Tab>

  <Tab title="Java">
    ```java theme={null}
    PulsarAdmin admin = // ...

    admin.namespaces().setSchemaCompatibilityStrategy("my-tenant/my-ns", SchemaCompatibilityStrategy.FULL);
    ```
  </Tab>
</Tabs>

To read the current value back, use the matching `get-schema-compatibility-strategy` subcommand at
either level.

## Manage AutoUpdate

[AutoUpdate](/cloud/governance/sr/pulsar/compatibility#schema-autoupdate) is enabled by default. It's
a namespace-level policy.

```bash theme={null}
# Enable
pulsarctl namespaces set-is-allow-auto-update-schema --enable <tenant>/<namespace>

# Disable
pulsarctl namespaces set-is-allow-auto-update-schema --disable <tenant>/<namespace>
```

Disable it when you want schemas registered only through an explicit
[upload](#upload-a-schema)—for example, when schema changes go through a review process rather than
being registered by whichever producer connects first.

## Manage schema validation

[Schema validation enforcement](/cloud/governance/sr/pulsar/compatibility#schema-validation-enforcement)
is disabled for producers by default, which means a producer carrying no schema can write raw bytes
to a topic that has one. Enable it to reject those producers. It's a namespace-level policy.

```bash theme={null}
# Enable
pulsarctl namespaces set-schema-validation-enforce --enable <tenant>/<namespace>

# Disable
pulsarctl namespaces set-schema-validation-enforce --disable <tenant>/<namespace>
```

## Create a schema in the Cloud Console

You can also attach a schema to a topic when you create it in the StreamNative Cloud Console. See
[Create schema for topics](/cloud/manage-data-streams/topic#create-schema-for-topics).

## What's next

<CardGroup cols={2}>
  <Card title="Compatibility" icon="code-compare" href="/cloud/governance/sr/pulsar/compatibility">
    Choose the right strategy before you change a schema.
  </Card>

  <Card title="Use with clients" icon="code" href="/cloud/governance/sr/pulsar/use-with-clients">
    Produce and consume typed messages.
  </Card>
</CardGroup>

<Note title="Attribution">
  Parts of this page are adapted from the [Apache Pulsar documentation](https://pulsar.apache.org/docs/admin-api-schemas/),
  licensed under the [Apache License 2.0](https://www.apache.org/licenses/LICENSE-2.0).
</Note>