Manage Schemas

tip

This page only shows some frequently used operations.

• For the latest and complete information about Pulsar admin, including commands, flags, descriptions, and more, see Pulsar admin docs.

• For the latest and complete information about REST API, including parameters, responses, samples, and more, see REST API doc.

• For the latest and complete information about Java admin API, including classes, methods, descriptions, and more, see Java admin API doc.

Manage schema

Upload a schema

To upload (register) a new schema for a topic, you can use one of the following methods.

Admin CLI

Use the upload subcommand.

pulsar-admin schemas upload --filename <schema-definition-file> <topic-name>

The schema-definition-file is in JSON format.

{
    "type": "<schema-type>",
    "schema": "<an-utf8-encoded-string-of-schema-definition-data>",
    "properties": {} // the properties associated with the schema
}

REST API

Send a POST request to the endpoint documented here: POST /admin/v2/schemas/{tenant}/{namespace}/{topic}/schema

Below is an example with CURL with a payload stored on the schema.json file, Pulsar broker running on localhost and the topic my-tenant/my-ns/my-topic:

curl -X POST -H 'Content-Type: application/json' -d @schema.json http://localhost:8080/admin/v2/schemas/my-tenant/my-ns/my-topic/schema

The post payload is in JSON format.

{
    "type": "<schema-type>",
    "schema": "<an-utf8-encoded-string-of-schema-definition-data>",
    "properties": {} // the properties associated with the schema
}

Java

The method on PulsarAdmin client is:

void createSchema(String topic, PostSchemaPayload schemaPayload)

Here is an example of PostSchemaPayload:

PulsarAdmin admin = …;

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

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

If the schema is a primitive schema, the schema field must be blank. If the schema is a struct schema, this field must be a JSON string of the Avro schema definition.

The payload includes the following fields:

Field	Description
type	Allowed values for primitive-type schemas are listed on the following page: Primitive types Allowed values for struct-type schemas are AVRO, PROTOBUF, PROTOBUF_NATIVE and JSON.
schema	The schema definition data, which is encoded in UTF 8 charset. If the schema type is AVRO, PROTOBUF or JSON schema, this field should be an Avro schema definition in JSON format. If the schema type is PROTOBUF_NATIVE schema, this field should contain a Protobuf descriptor. Otherwise, this field should be blank.
properties	The additional properties associated with the schema.

The following is an example for a 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 the latest schema

To get the latest schema for a topic, you can use one of the following methods.

Admin CLI

Use the get subcommand.

pulsar-admin schemas get <topic-name>

Pulsar preserves explicit JSON null values in schema output, including Avro field defaults such as "default": null. A null default and an omitted default have different schema semantics; preserve these values when inspecting or transferring a schema.

Example output:

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

REST API

Send a GET request to this endpoint: GET /admin/v2/schemas/{tenant}/{namespace}/{topic}/schema

Here is an example of a response, which is returned in JSON format.

{
    "version": "<the-version-number-of-the-schema>",
    "type": "<the-schema-type>",
    "timestamp": "<the-creation-timestamp-of-the-version-of-the-schema>",
    "data": "<an-utf8-encoded-string-of-schema-definition-data>",
    "properties": {} // the properties associated with the schema
}

Java

SchemaInfo createSchema(String topic)

Here is an example of SchemaInfo:

PulsarAdmin admin = …;

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

Get a specific schema

To get a specific version of a schema, you can use one of the following methods.

Admin CLI

Use the get subcommand.

pulsar-admin schemas get <topic-name> --version <version>

REST API

Send a GET request to a schema endpoint: GET /admin/v2/schemas/{tenant}/{namespace}/{topic}/schema/{version}

Here is an example of a response, which is returned in JSON format.

{
    "version": "<the-version-number-of-the-schema>",
    "type": "<the-schema-type>",
    "timestamp": "<the-creation-timestamp-of-the-version-of-the-schema>",
    "data": "<an-utf8-encoded-string-of-schema-definition-data>",
    "properties": {} // the properties associated with the schema
}

Java

SchemaInfo createSchema(String topic, long version)

Here is an example of SchemaInfo:

PulsarAdmin admin = …;

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

Extract a schema

To extract (provide) a schema via a topic, use the following method.

Admin CLI

Use the extract subcommand.

pulsar-admin schemas extract --classname <class-name> --jar <absolute-jar-path> --type <type-name>

Delete a schema

note

In any case, the delete action deletes all versions of a schema registered for a topic.

To delete a schema for a topic, you can use one of the following methods.

Admin CLI

Use the delete subcommand.

pulsar-admin schemas delete <topic-name>

REST API

Send a DELETE request to a schema endpoint: DELETE /admin/v2/schemas/{tenant}/{namespace}/{topic}/schema

Here is an example of a response returned in JSON format.

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

Java

void deleteSchema(String topic)

Here is an example of deleting a schema.

PulsarAdmin admin = …;

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

Manage schema AutoUpdate

Enable schema AutoUpdate

To enable/enforce schema auto-update at the namespace level, you can use one of the following methods.

Admin CLI

Use the set-is-allow-auto-update-schema subcommand.

bin/pulsar-admin namespaces set-is-allow-auto-update-schema --enable tenant/namespace

REST API

Send a POST request to a namespace endpoint: POST /admin/v2/namespaces/{tenant}/{namespace}/isAllowAutoUpdateSchema

The post payload is in JSON format.

true

Java

Here is an example to enable schema auto-update for a tenant/namespace.

admin.namespaces().setIsAllowAutoUpdateSchema("tenant/namespace", true, null);

Disable schema AutoUpdate

note

When schema auto-update is disabled, ordinary producers cannot register a new schema automatically; upload the schema explicitly. By default, geo-replication producers can still register compatible schemas. See Control schema registration by replicators.

To disable schema auto-update at the namespace level, you can use one of the following commands.

Admin CLI

Use the set-is-allow-auto-update-schema subcommand.

bin/pulsar-admin namespaces set-is-allow-auto-update-schema --disable tenant/namespace

REST API

Send a POST request to a namespace endpoint: POST /admin/v2/namespaces/{tenant}/{namespace}/isAllowAutoUpdateSchema

The post payload is in JSON format.

false

Java

Here is an example to disable schema auto-update for ordinary producers while leaving the replicator policy unchanged.

admin.namespaces().setIsAllowAutoUpdateSchema("tenant/namespace", false, null);

Control schema registration by replicators

Pulsar allows replication producers to register compatible schemas by default, even when automatic schema updates are disabled for ordinary producers. This lets the destination accept schemas used by replicated messages without also permitting local applications to register new schemas. Schema compatibility checks still apply.

To disable automatic registration for both ordinary producers and replicators:

bin/pulsar-admin namespaces set-is-allow-auto-update-schema \
  --disable --enable-for-replicator false tenant/namespace

Pre-register the required schemas at the destination before disabling replicator updates; otherwise replication can fail when it encounters an unregistered schema. To keep ordinary producer updates disabled but allow replicators:

bin/pulsar-admin namespaces set-is-allow-auto-update-schema \
  --disable --enable-for-replicator true tenant/namespace

Omitting --enable-for-replicator leaves that policy unchanged. Enabling auto-update for ordinary producers while explicitly disabling it for replicators is rejected.

The REST endpoint accepts a JSON boolean body for the ordinary producer setting and the optional query parameter allowAutoUpdateSchemaWithReplicator for the replicator setting. The Java admin API now takes three arguments:

admin.namespaces().setIsAllowAutoUpdateSchema("tenant/namespace", false, false);

The third argument is a nullable Boolean; null preserves the existing replicator setting. Applications compiled against the former two-argument method must update their calls and recompile, including calls to setIsAllowAutoUpdateSchemaAsync.

Manage schema validation enforcement

Enable schema validation enforcement

To enforce schema validation enforcement at the cluster level, you can configure isSchemaValidationEnforced to true in the conf/broker.conf file.

To enable schema validation enforcement at the namespace level, you can use one of the following commands.

Admin CLI

Use the set-schema-validation-enforce subcommand.

bin/pulsar-admin namespaces set-schema-validation-enforce --enable tenant/namespace

REST API

Send a POST request to a namespace endpoint: POST /admin/v2/namespaces/{tenant}/{namespace}/schemaValidationEnforced

The post payload is in JSON format.

{
"schemaValidationEnforced": "true"
}

Java

Here is an example to enable schema validation enforcement for a tenant/namespace.

admin.namespaces().setSchemaValidationEnforced("my-namspace", true);

Disable schema validation enforcement

To disable schema validation enforcement at the namespace level, you can use one of the following commands.

Admin CLI

Use the set-schema-validation-enforce subcommand.

bin/pulsar-admin namespaces set-schema-validation-enforce --disable tenant/namespace

REST API

Send a POST request to a namespace endpoint: POST /admin/v2/namespaces/{tenant}/{namespace}/schemaValidationEnforced

The post payload is in JSON format.

{
"schemaValidationEnforced": "false"
}

Java

Here is an example to enable schema validation enforcement for a tenant/namespace.

admin.namespaces().setSchemaValidationEnforced("my-namspace", false);

Manage schema compatibility strategy

The schema compatibility check strategy configured at different levels has priority: topic level > namespace level > cluster level. In other words:

• If you set the strategy at both topic and namespace levels, the topic-level strategy is used.
• If you set the strategy at both namespace and cluster levels, the namespace-level strategy is used.

Set schema compatibility strategy

Set topic-level schema compatibility strategy

To set a schema compatibility check strategy at the topic level, you can use one of the following methods.

Admin CLI

Use the pulsar-admin topicPolicies set-schema-compatibility-strategy command.

pulsar-admin topicPolicies set-schema-compatibility-strategy <strategy> <topicName>

REST API

Send a PUT request to this endpoint: PUT /admin/v2/persistent/{tenant}/{namespace}/{topic}/schemaCompatibilityStrategy

Java

void setSchemaCompatibilityStrategy(String topic, SchemaCompatibilityStrategy strategy)

Here is an example of setting a schema compatibility check strategy at the topic level.

PulsarAdmin admin = …;

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

Set namespace-level schema compatibility strategy

To set schema compatibility check strategy at the namespace level, you can use one of the following methods.

Admin CLI

Use the pulsar-admin namespaces set-schema-compatibility-strategy command.

pulsar-admin namespaces set-schema-compatibility-strategy options

REST API

Send a PUT request to this endpoint: PUT /admin/v2/namespaces/{tenant}/{namespace}/schemaCompatibilityStrategy

Java

Use the setSchemaCompatibilityStrategy method.

admin.namespaces().setSchemaCompatibilityStrategy("test", SchemaCompatibilityStrategy.FULL);

Set cluster-level schema compatibility strategy

To set schema compatibility check strategy at the cluster level, set schemaCompatibilityStrategy in the conf/broker.conf file.

The following is an example:

schemaCompatibilityStrategy=ALWAYS_INCOMPATIBLE

Get schema compatibility strategy

Get topic-level schema compatibility strategy

To get the topic-level schema compatibility check strategy, you can use one of the following methods.

Admin CLI

Use the pulsar-admin topicPolicies get-schema-compatibility-strategy command.

pulsar-admin topicPolicies get-schema-compatibility-strategy <topicName>

REST API

Send a GET request to this endpoint: GET /admin/v2/persistent/{tenant}/{namespace}/{topic}/schemaCompatibilityStrategy

Java

SchemaCompatibilityStrategy getSchemaCompatibilityStrategy(String topic, boolean applied)

Here is an example of getting the topic-level schema compatibility check strategy.

PulsarAdmin admin = …;

// get the current applied schema compatibility strategy
admin.topicPolicies().getSchemaCompatibilityStrategy("my-tenant/my-ns/my-topic", true);

// only get the schema compatibility strategy from topic policies
admin.topicPolicies().getSchemaCompatibilityStrategy("my-tenant/my-ns/my-topic", false);

Get namespace-level schema compatibility strategy

You can get schema compatibility check strategy at namespace level using one of the following methods.

Admin CLI

Use the pulsar-admin namespaces get-schema-compatibility-strategy command.

pulsar-admin namespaces get-schema-compatibility-strategy options

REST API

Send a GET request to this endpoint: GET /admin/v2/namespaces/{tenant}/{namespace}/schemaCompatibilityStrategy

Java

Use the getSchemaCompatibilityStrategy method.

admin.namespaces().getSchemaCompatibilityStrategy("test", SchemaCompatibilityStrategy.FULL);