Manage Pulsar Schemas

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.
pulsarctl installed and configured, if you use the CLI.

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.

Upload a schema

To register a new schema for a topic:

pulsarctl
REST API
Java

pulsarctl schemas upload --filename <schema-definition-file> <topic-name>

Send a POST request to the schema endpoint. This example posts a payload stored in schema.json:

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

PulsarAdmin admin = // ...

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

admin.createSchema("my-tenant/my-ns/my-topic", payload);

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

{
    "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. 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 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:

{
    "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

pulsarctl
REST API
Java

pulsarctl schemas get <topic-name>

Example output:

{
    "version": 0,
    "type": "String",
    "timestamp": 0,
    "data": "string",
    "properties": {
        "property1": "string",
        "property2": "string"
    }
}

curl -H "Authorization: Bearer $TOKEN" \
  https://<pulsar-cluster-http-url>/admin/v2/schemas/my-tenant/my-ns/my-topic/schema

PulsarAdmin admin = // ...

SchemaInfo si = admin.getSchema("my-tenant/my-ns/my-topic");

Get a specific version

pulsarctl
REST API
Java

pulsarctl schemas get <topic-name> --version <version>

curl -H "Authorization: Bearer $TOKEN" \
  https://<pulsar-cluster-http-url>/admin/v2/schemas/my-tenant/my-ns/my-topic/schema/1

PulsarAdmin admin = // ...

SchemaInfo si = admin.getSchema("my-tenant/my-ns/my-topic", 1L);

Delete a schema

Deleting a schema removes all versions registered for the topic. There’s no per-version delete and no way to recover the versions afterward.

pulsarctl
REST API
Java

pulsarctl schemas delete <topic-name>

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:

{
    "version": "<the-latest-version-number-of-the-schema>"
}

PulsarAdmin admin = // ...

admin.deleteSchema("my-tenant/my-ns/my-topic");

Set the compatibility strategy

The compatibility strategy 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.

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.

Topic level

pulsarctl
Java

pulsarctl topicPolicies set-schema-compatibility-strategy <strategy> <topic-name>

PulsarAdmin admin = // ...

admin.topicPolicies().setSchemaCompatibilityStrategy(
    "my-tenant/my-ns/my-topic", SchemaCompatibilityStrategy.ALWAYS_INCOMPATIBLE);

Namespace level

pulsarctl
Java

pulsarctl namespaces set-schema-compatibility-strategy --compatibility <strategy> <tenant>/<namespace>

PulsarAdmin admin = // ...

admin.namespaces().setSchemaCompatibilityStrategy("my-tenant/my-ns", SchemaCompatibilityStrategy.FULL);

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

Manage AutoUpdate

AutoUpdate is enabled by default. It’s a namespace-level policy.

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

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

What’s next

Compatibility

Choose the right strategy before you change a schema.

Use with clients

Produce and consume typed messages.

Parts of this page are adapted from the Apache Pulsar documentation, licensed under the Apache License 2.0.

Get Started

Clusters

Data Streams

Process

Connect

Lakehouse

Governance

Pulsar Clients

MQTT Clients

Tools

Pulsar Changelogs

Prerequisites

Upload a schema

Get a schema

Get the latest version

Get a specific version

Delete a schema

Set the compatibility strategy

Topic level

Namespace level

Manage AutoUpdate

Manage schema validation

Create a schema in the Cloud Console

What’s next

Compatibility

Use with clients

​Prerequisites

​Upload a schema

​Get a schema

​Get the latest version

​Get a specific version

​Delete a schema

​Set the compatibility strategy

​Topic level

​Namespace level

​Manage AutoUpdate

​Manage schema validation

​Create a schema in the Cloud Console

​What’s next

Compatibility

Use with clients

Prerequisites

Upload a schema

Get a schema

Get the latest version

Get a specific version

Delete a schema

Set the compatibility strategy

Topic level

Namespace level

Manage AutoUpdate

Manage schema validation

Create a schema in the Cloud Console

What’s next