Skip to main content

Manage schema

This guide demonstrates the ways to manage schemas:

Schema AutoUpdate​

If a schema passes the schema compatibility check, Pulsar producer automatically updates this schema to the topic it produces by default.

AutoUpdate for producer​

For a producer, the AutoUpdate happens in the following cases:

  • If a topic doesn’t have a schema, Pulsar registers a schema automatically.

  • If a topic has a schema:

    • If a producer doesn’t carry a schema:

    • If isSchemaValidationEnforced or schemaValidationEnforced is disabled in the namespace to which the topic belongs, the producer is allowed to connect to the topic and produce data.

    • If isSchemaValidationEnforced or schemaValidationEnforced is enabled in the namespace to which the topic belongs, the producer is rejected and disconnected.

    • If a producer carries a schema:

    A broker performs the compatibility check based on the configured compatibility check strategy of the namespace to which the topic belongs.

    • If it is a new schema and it passes the compatibility check, the broker registers a new schema automatically for the topic.

    • If the schema does not pass the compatibility check, the broker does not register a schema.

AutoUpdate Producer

AutoUpdate for consumer​

For a consumer, the AutoUpdate happens in the following cases:

  • If a consumer connects to a topic without a schema (which means the consumer receiving raw bytes), the consumer can connect to the topic successfully without doing any compatibility check.

  • If a consumer connects to a topic with a schema:

    • If the topic is idle (no producers, no entries, no other consumers and no registered schemas), the broker registers a schema for the topic automatically.

    • If the topic is not idle, the broker verifies if the schema provided by the consumer is compatible with the registered schema of the topic.

    • If the schema passes the compatibility check, the consumer can connect to the topic and receive messages.

    • If the schema does not pass the compatibility check, the consumer is rejected and disconnected.

AutoUpdate Producer

Manage AutoUpdate strategy​

You can use the pulsar-admin command to manage the AutoUpdate strategy as below:

Disable AutoUpdate​

To disable AutoUpdate on a namespace, you can use the pulsar-admin command.


bin/pulsar-admin namespaces set-schema-autoupdate-strategy --disabled tenant/namespace

Once the AutoUpdate is disabled, you can only register a new schema using the pulsar-admin command.

Adjust compatibility​

To adjust the schema compatibility level on a namespace, you can use the pulsar-admin command.


bin/pulsar-admin namespaces set-schema-autoupdate-strategy --compatibility <compatibility-level> tenant/namespace

Schema validation​

By default, schemaValidationEnforced is disabled for producers:

  • This means a producer without a schema can produce any kind of messages to a topic with schemas, which may result in producing trash data to the topic.

  • This allows non-java language clients that don’t support schema can produce messages to a topic with schemas.

However, if you want a stronger guarantee on the topics with schemas, you can enable schemaValidationEnforced across the whole cluster or on a per-namespace basis.

Enable schema validation​

To enable schemaValidationEnforced on a namespace, you can use the pulsar-admin command.


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

Disable schema validation​

To disable schemaValidationEnforced on a namespace, you can use the pulsar-admin command.


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

Schema manual management​

To manage schemas, you can use one of the following methods.

MethodDescription
Admin CLI
  • You can use the pulsar-admin tool to manage Pulsar schemas, brokers, clusters, sources, sinks, topics, tenants and so on. For more information about how to use the pulsar-admin tool, see here.
    REST API
  • Pulsar exposes schema related management API in Pulsar’s admin RESTful API. You can access the admin RESTful endpoint directly to manage schemas. For more information about how to use the Pulsar REST API, see here.
    Java Admin API
  • Pulsar provides Java admin library.

    Upload a schema​

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

    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
    }

    The schema-definition-file includes the following fields:

    FieldDescription
    typeThe schema type.
    schemaThe schema definition data, which is encoded in UTF 8 charset.
  • If the schema is a primitive schema, this field should be blank.
  • If the schema is a struct schema, this field should be a JSON string of the Avro schema definition.
  • propertiesThe additional properties associated with the schema.

    Here are examples of the schema-definition-file for a JSON schema.

    Example 1


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

    Example 2


    {
    "type": "STRING",
    "schema": "",
    "properties": {
    "key1": "value1"
    }
    }

    Get a schema (latest)​

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

    Use the get subcommand.


    $ pulsar-admin schemas get <topic-name>

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

    Get a schema (specific)​

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

    Use the get subcommand.


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

    Extract a schema​

    To provide a schema via a topic, you can use the following method.

    Use the extract subcommand.


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

    Delete a schema​

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

    note

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

    Use the delete subcommand.


    $ pulsar-admin schemas delete <topic-name>

    Custom schema storage​

    By default, Pulsar stores various data types of schemas in Apache BookKeeper deployed alongside Pulsar.

    However, you can use another storage system if needed.

    Implement​

    To use a non-default (non-BookKeeper) storage system for Pulsar schemas, you need to implement the following Java interfaces:

    SchemaStorage interface​

    The SchemaStorage interface has the following methods:


    public interface SchemaStorage {
    // How schemas are updated
    CompletableFuture<SchemaVersion> put(String key, byte[] value, byte[] hash);

    // How schemas are fetched from storage
    CompletableFuture<StoredSchema> get(String key, SchemaVersion version);

    // How schemas are deleted
    CompletableFuture<SchemaVersion> delete(String key);

    // Utility method for converting a schema version byte array to a SchemaVersion object
    SchemaVersion versionFromBytes(byte[] version);

    // Startup behavior for the schema storage client
    void start() throws Exception;

    // Shutdown behavior for the schema storage client
    void close() throws Exception;
    }

    tip

    For a complete example of schema storage implementation, see BookKeeperSchemaStorage class.

    SchemaStorageFactory interface​

    The SchemaStorageFactory interface has the following method:


    public interface SchemaStorageFactory {
    @NotNull
    SchemaStorage create(PulsarService pulsar) throws Exception;
    }

    tip

    For a complete example of schema storage factory implementation, see BookKeeperSchemaStorageFactory class.

    Deploy​

    To use your custom schema storage implementation, perform the following steps.

    1. Package the implementation in a JAR file.

    2. Add the JAR file to the lib folder in your Pulsar binary or source distribution.

    3. Change the schemaRegistryStorageClassName configuration in broker.conf to your custom factory class.

    4. Start Pulsar.