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>

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.

{
"isAllowAutoUpdateSchema": "true"
}

Java

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

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

Disable schema AutoUpdate

note

When schema auto-update is disabled, you can only register a new schema.

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.

{
"isAllowAutoUpdateSchema": "false"
}

Java

Here is an example to enable schema auto-unpdate of a tenant/namespace.

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

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);