> ## Documentation Index
> Fetch the complete documentation index at: https://docs.streamnative.io/llms.txt
> Use this file to discover all available pages before exploring further.

# Use Pulsar Schemas with Clients

> Construct each Pulsar schema type in your application and use it to produce and consume typed messages.

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](/cloud/governance/sr/pulsar/overview#how-it-works).

## Prerequisites

* A Pulsar cluster on StreamNative Cloud, and a client configured to connect to it. See
  [Connect to your cluster](/cloud/build/pulsar-clients/qs-connect).
* A service account with `produce` and `consume` permissions on the target topic.

## Primitive schemas

### bytes

<Tabs>
  <Tab title="Java">
    ```java theme={null}
    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);
    ```
  </Tab>

  <Tab title="Python">
    ```python theme={null}
    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()
    ```
  </Tab>

  <Tab title="Go">
    ```go theme={null}
    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,
    })
    ```
  </Tab>

  <Tab title="C++">
    ```cpp theme={null}
    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);
    ```
  </Tab>
</Tabs>

### string

<Tabs>
  <Tab title="Java">
    ```java theme={null}
    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();
    ```
  </Tab>

  <Tab title="Python">
    ```python theme={null}
    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()
    ```
  </Tab>

  <Tab title="Go">
    ```go theme={null}
    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())
    ```
  </Tab>

  <Tab title="C++">
    ```cpp theme={null}
    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);
    ```
  </Tab>
</Tabs>

The other primitive types follow the same pattern. See
[Primitive types](/cloud/governance/sr/pulsar/schema-types#primitive-types) for the full list and the
language-specific type each one maps to.

## Key/value schemas

A [key/value schema](/cloud/governance/sr/pulsar/schema-types#keyvalue-schema) pairs a schema for the
key with a schema for the value.

1. 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.

   ```java theme={null}
   Schema<KeyValue<Integer, String>> kvSchema = Schema.KeyValue(
       Schema.INT32,
       Schema.STRING,
       KeyValueEncodingType.INLINE
   );
   ```

2. Produce with it:

   ```java theme={null}
   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();
   ```

3. Consume with it:

   ```java theme={null}
   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.

```java theme={null}
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;
    }
}
```

```java theme={null}
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:

```cpp theme={null}
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`:

```java theme={null}
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

1. Generate the message class with Protobuf 3 or later:

   ```protobuf theme={null}
   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;
   }
   ```

2. Build a producer and consumer with `Schema.PROTOBUF`:

   ```java theme={null}
   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.

```java theme={null}
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.

```java theme={null}
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`.

```java theme={null}
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

<CardGroup cols={2}>
  <Card title="Schema types" icon="shapes" href="/cloud/governance/sr/pulsar/schema-types">
    The full type reference and per-language mappings.
  </Card>

  <Card title="Manage schemas" icon="sliders" href="/cloud/governance/sr/pulsar/manage-schemas">
    Upload, retrieve, and delete schemas outside your application.
  </Card>
</CardGroup>

<Note title="Attribution">
  Parts of this page are adapted from the [Apache Pulsar documentation](https://pulsar.apache.org/docs/schema-get-started/),
  licensed under the [Apache License 2.0](https://www.apache.org/licenses/LICENSE-2.0).
</Note>