Use Pulsar Schemas with Clients

Construct a schema in your client and pass it when you build a producer or consumer. The client handles serialization on the way out and deserialization on the way in, and the broker registers or validates the schema as described in How it works.

Prerequisites

A Pulsar cluster on StreamNative Cloud, and a client configured to connect to it. See Connect to your cluster.
A service account with produce and consume permissions on the target topic.

Primitive schemas

bytes

Java
Python
Go
C++

Producer<byte[]> producer = pulsarClient.newProducer(Schema.BYTES)
       .topic("my-topic")
       .create();
Consumer<byte[]> consumer = pulsarClient.newConsumer(Schema.BYTES)
       .topic("my-topic")
       .subscriptionName("my-sub")
       .subscribe();

producer.newMessage().value("message".getBytes()).send();

Message<byte[]> message = consumer.receive(5, TimeUnit.SECONDS);

producer = client.create_producer(
    'bytes-schema-topic',
    schema=BytesSchema())
producer.send(b"Hello")

consumer = client.subscribe(
    'bytes-schema-topic',
    'sub',
    schema=BytesSchema())
msg = consumer.receive()
data = msg.value()

producer, err := client.CreateProducer(pulsar.ProducerOptions{
    Topic:  "my-topic",
    Schema: pulsar.NewBytesSchema(nil),
})
id, err := producer.Send(context.Background(), &pulsar.ProducerMessage{
    Value: []byte("message"),
})

consumer, err := client.Subscribe(pulsar.ConsumerOptions{
    Topic:            "my-topic",
    Schema:           pulsar.NewBytesSchema(nil),
    SubscriptionName: "my-sub",
    Type:             pulsar.Exclusive,
})

SchemaInfo schemaInfo = SchemaInfo(SchemaType::BYTES, "Bytes", "");
Producer producer;
client.createProducer("topic-bytes", ProducerConfiguration().setSchema(schemaInfo), producer);
std::array<char, 1024> buffer;
producer.send(MessageBuilder().setContent(buffer.data(), buffer.size()).build());

Consumer consumer;
client.subscribe("topic-bytes", "my-sub", ConsumerConfiguration().setSchema(schemaInfo), consumer);
Message msg;
consumer.receive(msg, 3000);

string

Java
Python
Go
C++

Producer<String> producer = client.newProducer(Schema.STRING).create();
producer.newMessage().value("Hello Pulsar!").send();

Consumer<String> consumer = client.newConsumer(Schema.STRING).subscribe();
Message<String> message = consumer.receive();

producer = client.create_producer(
    'string-schema-topic',
    schema=StringSchema())
producer.send("Hello")

consumer = client.subscribe(
    'string-schema-topic',
    'sub',
    schema=StringSchema())
msg = consumer.receive()
value = msg.value()

producer, err := client.CreateProducer(pulsar.ProducerOptions{
    Topic:  "my-topic",
    Schema: pulsar.NewStringSchema(nil),
})
id, err := producer.Send(context.Background(), &pulsar.ProducerMessage{
    Value: "message",
})

consumer, err := client.Subscribe(pulsar.ConsumerOptions{
    Topic:            "my-topic",
    Schema:           pulsar.NewStringSchema(nil),
    SubscriptionName: "my-sub",
    Type:             pulsar.Exclusive,
})
msg, err := consumer.Receive(context.Background())

SchemaInfo schemaInfo = SchemaInfo(SchemaType::STRING, "String", "");
Producer producer;
client.createProducer("topic-string", ProducerConfiguration().setSchema(schemaInfo), producer);
producer.send(MessageBuilder().setContent("message").build());

Consumer consumer;
client.subscribe("topic-string", "my-sub", ConsumerConfiguration().setSchema(schemaInfo), consumer);
Message msg;
consumer.receive(msg, 3000);

The other primitive types follow the same pattern. See Primitive types for the full list and the language-specific type each one maps to.

Key/value schemas

A key/value schema pairs a schema for the key with a schema for the value.

Construct the schema, choosing an encoding type. INLINE puts both key and value in the message payload; SEPARATED stores the key as the message key and the value as the payload.
Schema<KeyValue<Integer, String>> kvSchema = Schema.KeyValue( Schema.INT32, Schema.STRING, KeyValueEncodingType.INLINE );

Produce with it:

Producer<KeyValue<Integer, String>> producer = client.newProducer(kvSchema)
    .topic(topicName)
    .create();

final int key = 100;
final String value = "value-100";

producer.newMessage()
    .value(new KeyValue(key, value))
    .send();

Consume with it:

Consumer<KeyValue<Integer, String>> consumer = client.newConsumer(kvSchema)
    .topic(topicName)
    .subscriptionName(subscriptionName)
    .subscribe();

Struct schemas

Avro

Pass a class and Pulsar extracts the schema definition from it.

public class SensorReading {
    public float temperature;

    public SensorReading(float temperature) {
        this.temperature = temperature;
    }

    // A no-arg constructor is required
    public SensorReading() {
    }

    public float getTemperature() {
        return temperature;
    }

    public void setTemperature(float temperature) {
        this.temperature = temperature;
    }
}

Producer<SensorReading> producer = client.newProducer(AvroSchema.of(SensorReading.class))
        .topic("sensor-readings")
        .create();

In C++, Go, and Python, pass the Avro schema definition as a JSON string instead:

static const std::string exampleSchema =
    "{\"type\":\"record\",\"name\":\"Example\",\"namespace\":\"test\","
    "\"fields\":[{\"name\":\"a\",\"type\":\"int\"},{\"name\":\"b\",\"type\":\"int\"}]}";
ProducerConfiguration producerConf;
producerConf.setSchema(SchemaInfo(AVRO, "Avro", exampleSchema));
client.createProducer("topic-avro", producerConf, producer);

JSON

Declaring a JSON schema works the same way as Avro—use Schema.JSON instead of AvroSchema:

static class SchemaDemo {
   public String name;
   public int age;
}

Producer<SchemaDemo> producer = pulsarClient.newProducer(Schema.JSON(SchemaDemo.class))
       .topic("my-topic")
       .create();
Consumer<SchemaDemo> consumer = pulsarClient.newConsumer(Schema.JSON(SchemaDemo.class))
       .topic("my-topic")
       .subscriptionName("my-sub")
       .subscribe();

SchemaDemo schemaDemo = new SchemaDemo();
schemaDemo.name = "pulsar";
schemaDemo.age = 20;
producer.newMessage().value(schemaDemo).send();

Message<SchemaDemo> message = consumer.receive(5, TimeUnit.SECONDS);

Protobuf

Generate the message class with Protobuf 3 or later:

syntax = "proto3";
message DemoMessage {
   string stringField = 1;
   double doubleField = 2;
   int32 intField = 6;
   TestEnum testEnum = 4;
   SubMessage nestedField = 5;
   repeated string repeatedField = 10;
   proto.external.ExternalMessage externalMessage = 11;
}

Build a producer and consumer with Schema.PROTOBUF:

Producer<DemoMessage> producer = pulsarClient.newProducer(Schema.PROTOBUF(DemoMessage.class))
       .topic("my-topic")
       .create();
Consumer<DemoMessage> consumer = pulsarClient.newConsumer(Schema.PROTOBUF(DemoMessage.class))
       .topic("my-topic")
       .subscriptionName("my-sub")
       .subscribe();

producer.newMessage().value(DemoMessage.newBuilder()
    .setStringField("string-field-value")
    .setIntField(1)
    .build()).send();

Message<DemoMessage> message = consumer.receive(5, TimeUnit.SECONDS);

ProtobufNative

ProtobufNative uses the Protobuf native descriptor rather than converting to Avro. Construct it exactly as above, substituting Schema.PROTOBUF_NATIVE for Schema.PROTOBUF. Use it when you want native protobuf-v3 serialization or AUTO_CONSUME support.

Native Avro

NATIVE_AVRO wraps an existing org.apache.avro.Schema and accepts an already-serialized Avro payload without re-validating it—useful when you’re ingesting data that another system already validated.

org.apache.avro.Schema nativeAvroSchema = // ...
Producer<byte[]> producer = pulsarClient.newProducer().topic("ingress").create();
byte[] content = // ...
producer.newMessage(Schema.NATIVE_AVRO(nativeAvroSchema)).value(content).send();

Auto schemas

AUTO_PRODUCE

Use AUTO_PRODUCE_BYTES when your application forwards already-serialized bytes into a topic that has a schema, and you want Pulsar to verify those bytes are compatible before they land.

Producer<byte[]> pulsarProducer = client.newProducer(Schema.AUTO_PRODUCE_BYTES())
    // ...
    .create();
byte[] sourceMessageBytes = // ...
pulsarProducer.send(sourceMessageBytes);

AUTO_CONSUME

Use AUTO_CONSUME when you don’t know the topic’s schema in advance. The client retrieves the SchemaInfo from the broker and deserializes each message into a GenericRecord.

Consumer<GenericRecord> pulsarConsumer = client.newConsumer(Schema.AUTO_CONSUME())
    // ...
    .subscribe();

Message<GenericRecord> msg = pulsarConsumer.receive();
GenericRecord record = msg.getValue();
record.getFields().forEach(field -> {
   if (field.getName().equals("theFieldYouNeed")) {
       Object recordField = record.getField(field);
       // Handle the field
   }
});

What’s next

Schema types

The full type reference and per-language mappings.

Manage schemas

Upload, retrieve, and delete schemas outside your application.

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

Primitive schemas

bytes

string

Key/value schemas

Struct schemas

Avro

JSON

Protobuf

ProtobufNative

Native Avro

Auto schemas

AUTO_PRODUCE

AUTO_CONSUME

What’s next

Schema types

Manage schemas

​Prerequisites

​Primitive schemas

​bytes

​string

​Key/value schemas

​Struct schemas

​Avro

​JSON

​Protobuf

​ProtobufNative

​Native Avro

​Auto schemas

​AUTO_PRODUCE

​AUTO_CONSUME

​What’s next

Schema types

Manage schemas

Prerequisites

Primitive schemas

bytes

string

Key/value schemas

Struct schemas

Avro

JSON

Protobuf

ProtobufNative

Native Avro

Auto schemas

AUTO_PRODUCE

AUTO_CONSUME

What’s next