Understand schema

This section explains the basic concepts of Pulsar schema and provides additional references.

Schema definition

Pulsar schema is defined in a data structure called SchemaInfo. It is stored and enforced on a per-topic basis and cannot be stored at the namespace or tenant level.

This is a string example of SchemaInfo.

{
    "name": "test-string-schema",
    "type": "STRING",
    "schema": "",
    "properties": {}
}

The following table outlines the fields that each SchemaInfo consists of.

Schema type

Pulsar supports various schema types, which are mainly divided into two categories:

• Primitive type
• Complex type

Primitive type

The following table outlines the primitive types that Pulsar schema supports, and the conversions between schema types and language-specific primitive types.

note

Pulsar does not store any schema data in SchemaInfo for primitive types. Some of the primitive schema implementations can use the properties parameter to store implementation-specific tunable settings. For example, a string schema can use properties to store the encoding charset to serialize and deserialize strings.

For more instructions and examples, see Construct a string schema.

Complex type

The following table outlines the complex types that Pulsar schema supports:

KeyValue schema

KeyValue schema helps applications define schemas for both key and value. Pulsar stores the SchemaInfo of the key schema and the value schema together.

Pulsar provides the following methods to encode a single key/value pair in a message：

• INLINE - Key/Value pairs are encoded together in the message payload.
• SEPARATED - The Key is stored as a message key, while the value is stored as the message payload. See Construct a key/value schema for more details.

Struct schema

The following table outlines the struct types that Pulsar schema supports:

Pulsar provides the following methods to use the struct schema.

• static
• generic
• SchemaDefinition

This example shows how to construct a struct schema with these methods and use it to produce and consume messages.

• static
• generic
• SchemaDefinition

You can predefine the struct schema, which can be a POJO in Java, a struct in Go, or classes generated by Avro or Protobuf tools.

Example

Pulsar gets the schema definition from the predefined struct using an Avro library. The schema definition is the schema data stored as a part of the SchemaInfo.

1. Create the User class to define the messages sent to Pulsar topics.

   # If you use Lombok
   @Builder
   @AllArgsConstructor
   @NoArgsConstructor
   public static class User {
       public String name;
       public int age;
   }
   # If you DON'T use Lombok you will need to add the constructor like this
   #
   #   public static class User {
   #    String name;
   #    int age;
   #    public User() { }
   #    public User(String name, int age) { this.name = name; this.age = age; } }
   #}

2. Create a producer with a struct schema and send messages.

   Producer<User> producer = client.newProducer(Schema.AVRO(User.class)).create();
   producer.newMessage().value(new User("pulsar-user", 1)).send();

3. Create a consumer with a struct schema and receive messages

   Consumer<User> consumer = client.newConsumer(Schema.AVRO(User.class)).subscribe();
   User user = consumer.receive().getValue();

Sometimes applications do not have pre-defined structs, and you can use this method to define schema and access data.

You can define the struct schema using the GenericSchemaBuilder, generate a generic struct using GenericRecordBuilder, and consume messages into GenericRecord.

Example

1. Use RecordSchemaBuilder to build a schema.

   RecordSchemaBuilder recordSchemaBuilder = SchemaBuilder.record("schemaName");
   recordSchemaBuilder.field("intField").type(SchemaType.INT32);
   SchemaInfo schemaInfo = recordSchemaBuilder.build(SchemaType.AVRO);
   Consumer<GenericRecord> consumer = client.newConsumer(Schema.generic(schemaInfo))
        .topic(topicName)
        .subscriptionName(subscriptionName)
        .subscribe();
   Producer<GenericRecord> producer = client.newProducer(Schema.generic(schemaInfo))
        .topic(topicName)
        .create();

2. Use RecordBuilder to build the struct records.

   GenericSchemaImpl schema = GenericAvroSchema.of(schemaInfo);
   // send message
   GenericRecord record = schema.newRecordBuilder().set("intField", 32).build();
   producer.newMessage().value(record).send();
   // receive message
   Message<GenericRecord> msg = consumer.receive();
   Assert.assertEquals(msg.getValue().getField("intField"), 32);

You can define the schemaDefinition to generate a struct schema.

Example

1. Create the User class to define the messages sent to Pulsar topics.

   public static class User {
       public String name;
       public int age;
       public User(String name, int age) {
   this.name = name;
   this.age = age
       }
       public User() {}
   }

2. Create a producer with a SchemaDefinition and send messages.

   SchemaDefinition<User> schemaDefinition = SchemaDefinition.<User>builder().withPojo(User.class).build();
   Producer<User> producer = client.newProducer(Schema.AVRO(schemaDefinition)).create();
   producer.newMessage().value(new User ("pulsar-user", 1)).send();

3. Create a consumer with a SchemaDefinition schema and receive messages.

   SchemaDefinition<User> schemaDefinition = SchemaDefinition.<User>builder().withPojo(User.class).build();
   Consumer<User> consumer = client.newConsumer(Schema.AVRO(schemaDefinition)).subscribe();
   User user = consumer.receive().getValue();

Auto Schema

If there is no chance to know the schema type of a Pulsar topic in advance, you can use AUTO schemas to produce/consume generic records to/from brokers.

Auto schema contains two categories:

• AUTO_PRODUCE transfers data from a producer to a Pulsar topic that has a schema and helps the producer validate whether the outbound bytes are compatible with the schema of the topic. For more instructions, see Construct an AUTO_PRODUCE schema.
• AUTO_CONSUME transfers data from a Pulsar topic that has a schema to a consumer and helps the topic validate whether the out-bound bytes are compatible with the consumer. In other words, the topic deserializes messages into language-specific objects GenericRecord using the SchemaInfo retrieved from brokers. For more instructions, see Construct an AUTO_CONSUME schema.

Schema validation enforcement

Schema validation enforcement enables brokers to reject producers/consumers without a schema.

By default, schema validation enforcement is only disabled (isSchemaValidationEnforced\=false) for producers, which means:

• A producer without a schema can produce any messages to a topic with schemas, which may result in producing trash data to the topic.
• Clients that don’t support schema are allowed to produce messages to a topic with schemas.

For how to enable schema validation enforcement, see Manage schema validation.

Schema evolution

Schemas store the details of attributes and types. To satisfy new business needs, schemas undergo evolution over time with versioning.

note

Schema evolution only applies to Avro, JSON, Protobuf, and ProtobufNative schemas.

Schema evolution may impact existing consumers. The following control measures have been designed to serve schema evolution and ensure the downstream consumers can seamlessly handle schema evolution:

• Schema compatibility check
• Schema AutoUpdate

For further readings about schema evolution, see Avro documentation and Protobuf documentation.

Schema versioning

Each SchemaInfo stored with a topic has a version. The schema version manages schema changes happening within a topic.

Messages produced with SchemaInfo are tagged with a schema version. When a message is consumed by a Pulsar client, the client can use the schema version to retrieve the corresponding SchemaInfo and use the correct schema to deserialize data. Once a version is assigned to or fetched from a schema, all subsequent messages produced by that producer are tagged with the appropriate version.

Suppose you are using a Pulsar Java client to create a producer and send messages.

PulsarClient client = PulsarClient.builder()
        .serviceUrl("pulsar://localhost:6650")
        .build();
Producer<SensorReading> producer = client.newProducer(JSONSchema.of(SensorReading.class))
        .topic("sensor-data")
        .sendTimeout(3, TimeUnit.SECONDS)
        .create();

The table below outlines the possible scenarios when this connection attempt occurs and the result of each scenario:

Schema compatibility check

The purpose of schema compatibility check is to ensure that existing consumers can process the introduced messages.

When receiving a SchemaInfo from producers, brokers recognize the schema type and deploy the schema compatibility checker (schemaRegistryCompatibilityCheckers) for that schema type to check if the SchemaInfo is compatible with the schema of the topic by applying the configured compatibility check strategy.

The default value of schemaRegistryCompatibilityCheckers in the conf/broker.conf file is as follows.

schemaRegistryCompatibilityCheckers=org.apache.pulsar.broker.service.schema.JsonSchemaCompatibilityCheck,org.apache.pulsar.broker.service.schema.AvroSchemaCompatibilityCheck,org.apache.pulsar.broker.service.schema.ProtobufNativeSchemaCompatibilityCheck

Each schema type corresponds to one instance of the schema compatibility checker. Avro, JSON, and Protobuf schemas have their own compatibility checkers, while all the other schema types share the default compatibility checker that disables the schema evolution.

Schema compatibility check strategy

Suppose that you have a topic containing three schemas (V1, V2, and V3). V1 is the oldest and V3 is the latest. The following table outlines 8 schema compatibility strategies and how it works.

tip

• The default schema compatibility check strategy varies depending on schema types.
  • For Avro and JSON, the default one is FULL.
  • For others, the default one is ALWAYS_INCOMPATIBLE.
• For more instructions about how to set the strategy, see Manage schemas.

Schema AutoUpdate

By default, schema AutoUpdate is enabled. When a schema passes the schema compatibility check, the producer automatically updates this schema to the topic it produces.

Producer side

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

• If a topic doesn’t have a schema (meaning the data is in raw bytes), Pulsar registers the schema automatically.

• If a topic has a schema and the producer doesn’t carry any schema (meaning it produces raw bytes):

  • If schema validation enforcement is disabled (schemaValidationEnforced\=false) in the namespace that the topic belongs to, the producer is allowed to connect to the topic and produce data.

  • Otherwise, the producer is rejected.

  • If a topic has a schema and the producer carries a schema, see How schema works on producer side for more information.

Consumer side

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

• If a consumer connects to a topic without a schema (meaning it consumes 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, see How schema works on consumer side for more information.

Order of upgrading clients

To adapt to schema evolution and auto-update, you need to upgrade your client applications accordingly. The upgrade order may vary depending on the configured schema compatibility check strategy.

The following table outlines the mapping between the schema compatibility check strategy and the upgrade order of clients.