> ## 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 External Protobuf Schema with Pulsar clients

> Use Kafka Protobuf Schema and the Kafka Schema Registry from Pulsar Java clients with the kafka-schemas library.

External Protobuf Schema lets Pulsar Java clients produce and consume messages that use [Kafka Protobuf Schema](/cloud/governance/kafka-schemas/kafka-schema-registry) and the [Kafka Schema Registry](/cloud/governance/kafka-schemas/kafka-schema-registry) on StreamNative Cloud. Schemas are registered in and resolved from the Kafka Schema Registry, while your application uses the familiar Pulsar `Producer` and `Consumer` APIs.

Use External Protobuf Schema when you want to:

* Use Pulsar clients with Kafka Protobuf Schema and Schema Registry compatibility checks.
* Share Protobuf schemas between Kafka and Pulsar clients on the same topic.
* Work with Protobuf message classes generated from `.proto` files.

The [`kafka-schemas`](https://github.com/streamnative/external-schemas) library provides a Pulsar `Schema` implementation backed by the Kafka Protobuf serializer. The same library also supports [External JSON Schema](/cloud/governance/kafka-schemas/external-json-schema) and [External Avro Schema](/cloud/governance/kafka-schemas/external-avro-schema).

## Prerequisites

* A StreamNative Pulsar cluster for message production and consumption.
* The [Kafka Schema Registry](/cloud/governance/kafka-schemas/kafka-schema-registry) service enabled on the cluster.
* A service account with `produce` and `consume` permissions on the target topic.
* RBAC permissions for the Kafka Schema Registry: assign the [`schema-writer`](/cloud/security/access/rbac/manage-rbac-roles#schema-writer) role to register schemas and the [`schema-reader`](/cloud/security/access/rbac/manage-rbac-roles#schema-reader) role to read schemas. See [Schema Registry RBAC roles](/cloud/security/access/rbac/manage-rbac-roles#schema-registry).
* Java 17 or higher.
* Pulsar Java client 4.1.0 or higher.

## Add the dependency

Add the following Maven dependencies to your project:

```xml theme={null}
<dependency>
  <groupId>org.apache.pulsar</groupId>
  <artifactId>pulsar-client</artifactId>
  <version>4.1.0</version>
  <exclusions>
      <exclusion>
          <groupId>javax.validation</groupId>
          <artifactId>validation-api</artifactId>
      </exclusion>
  </exclusions>
</dependency>
<dependency>
  <groupId>io.streamnative.schemas.external</groupId>
  <artifactId>kafka-schemas</artifactId>
  <version>1.0.0</version>
</dependency>
<dependency>
  <groupId>io.confluent</groupId>
  <artifactId>kafka-protobuf-serializer</artifactId>
  <version>8.0.0</version>
</dependency>
<dependency>
  <groupId>com.google.protobuf</groupId>
  <artifactId>protobuf-java</artifactId>
  <version>4.29.5</version>
</dependency>
```

The `pulsar-client` dependency provides the Pulsar `Producer`, `Consumer`, and `PulsarClient` APIs used in the examples below. Declare `kafka-protobuf-serializer` and `protobuf-java` explicitly so you control their versions. `kafka-schemas` also pulls them in transitively.

### Add the Kafka Maven repository

`pulsar-client`, `kafka-schemas`, and `protobuf-java` are available from [Maven Central](https://repo1.maven.org/maven2/). You do not need to add a repository for those dependencies.

`kafka-protobuf-serializer` is not published to Maven Central. Add the following repository to your `pom.xml`:

```xml theme={null}
<repositories>
  <repository>
    <id>kafka</id>
    <url>https://packages.confluent.io/maven/</url>
  </repository>
</repositories>
```

<Note>
  If your organization already mirrors `kafka-protobuf-serializer` artifacts in an internal repository, configure that mirror instead of adding the public repository directly.
</Note>

## Define a Protobuf schema

External Protobuf Schema works with Protobuf message classes generated from `.proto` files.

### Step 1: Create Protobuf definition files

Create `src/main/proto/other.proto`:

```protobuf theme={null}
syntax = "proto3";
package com.example.protobuf;

option java_multiple_files = true;
option java_package = "com.example.protobuf";

message OtherRecord {
  int32 other_id = 1;
}
```

Create `src/main/proto/myRecord.proto`:

```protobuf theme={null}
syntax = "proto3";
package com.example.protobuf;

option java_multiple_files = true;
option java_package = "com.example.protobuf";

import "other.proto";

message MyRecord {
  string f1 = 1;
  OtherRecord f2 = 2;
}
```

### Step 2: Generate the Protobuf classes

Add a Protobuf Maven plugin to your `pom.xml`. The following example uses the `protobuf-maven-plugin`:

```xml theme={null}
<plugin>
  <groupId>io.github.ascopes</groupId>
  <artifactId>protobuf-maven-plugin</artifactId>
  <version>3.10.1</version>
  <configuration>
    <protocVersion>4.29.5</protocVersion>
  </configuration>
  <executions>
    <execution>
      <goals>
        <goal>generate</goal>
      </goals>
    </execution>
  </executions>
</plugin>
```

Run `mvn generate-sources` to generate the `MyRecord` and `OtherRecord` classes in the `com.example.protobuf` package.

## Configure Schema Registry authentication

`KafkaSchemaFactory` accepts the same Schema Registry configuration properties as `KafkaProtobufSerializerConfig`. The [`external-schemas`](https://github.com/streamnative/external-schemas) examples authenticate to the Schema Registry with Basic authentication. Define the helper method as `private static` so you can call it from `main`.

Use your service account API key as the password. The username can be any non-empty string.

```java theme={null}
private static Map<String, Object> getSchemaRegistryConfigs(String schemaRegistryUrl, String apiKey) {
    Map<String, Object> configs = new HashMap<>();
    configs.put(KafkaProtobufSerializerConfig.SCHEMA_REGISTRY_URL_CONFIG, schemaRegistryUrl);
    configs.put(KafkaProtobufSerializerConfig.BASIC_AUTH_CREDENTIALS_SOURCE, "USER_INFO");
    configs.put(
            KafkaProtobufSerializerConfig.USER_INFO_CONFIG,
            String.format("%s:%s", "public", apiKey));
    return configs;
}
```

For additional serializer options, see the `KafkaProtobufSerializerConfig` class in the `kafka-protobuf-serializer` dependency.

## Produce and consume messages

Use `KafkaSchemaFactory` to create a Pulsar `Schema` backed by Kafka Protobuf Schema, then create a producer and consumer with the same schema instance.

```java theme={null}
import com.example.protobuf.MyRecord;
import com.example.protobuf.OtherRecord;
import io.confluent.kafka.serializers.protobuf.KafkaProtobufSerializerConfig;
import io.streamnative.schemas.external.KafkaSchemaFactory;
import java.util.HashMap;
import java.util.Map;
import org.apache.pulsar.client.api.AuthenticationFactory;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.SubscriptionInitialPosition;

public class ExternalProtobufSchemaExample {

    public static void main(String[] args) throws Exception {
        String serviceUrl = "<YOUR-PULSAR-SERVICE-URL>";
        String schemaRegistryUrl = "<YOUR-SCHEMA-REGISTRY-URL>";
        String apiKey = "<YOUR-API-KEY>";
        String topic = "persistent://public/default/protobuf-records";

        KafkaSchemaFactory schemaFactory =
                new KafkaSchemaFactory(getSchemaRegistryConfigs(schemaRegistryUrl, apiKey));
        Schema<MyRecord> schema = schemaFactory.protobuf(MyRecord.class);

        PulsarClient client = PulsarClient.builder()
                .serviceUrl(serviceUrl)
                .authentication(AuthenticationFactory.token(apiKey))
                .build();

        Producer<MyRecord> producer = client.newProducer(schema).topic(topic).create();
        Consumer<MyRecord> consumer = client.newConsumer(schema)
                .topic(topic)
                .subscriptionName("my-subscription")
                .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
                .subscribe();

        for (int i = 0; i < 10; i++) {
            MyRecord myRecord = MyRecord.newBuilder()
                    .setF1("name-" + i)
                    .setF2(OtherRecord.newBuilder().setOtherId(i).build())
                    .build();
            producer.send(myRecord);
        }

        for (int i = 0; i < 10; i++) {
            Message<MyRecord> message = consumer.receive();
            consumer.acknowledge(message);
            MyRecord myRecord = message.getValue();
            System.out.println("f1=>" + myRecord.getF1()
                    + ", f2.otherId=>" + myRecord.getF2().getOtherId());
        }

        consumer.close();
        producer.close();
        client.close();
    }

    private static Map<String, Object> getSchemaRegistryConfigs(String schemaRegistryUrl, String apiKey) {
        Map<String, Object> configs = new HashMap<>();
        configs.put(KafkaProtobufSerializerConfig.SCHEMA_REGISTRY_URL_CONFIG, schemaRegistryUrl);
        configs.put(KafkaProtobufSerializerConfig.BASIC_AUTH_CREDENTIALS_SOURCE, "USER_INFO");
        configs.put(
                KafkaProtobufSerializerConfig.USER_INFO_CONFIG,
                String.format("%s:%s", "public", apiKey));
        return configs;
    }
}
```

When the producer sends the first message, the schema is automatically registered in the Kafka Schema Registry. The consumer resolves the schema from the registry when reading messages.

## Schema compatibility

External Protobuf Schema registers schemas with the Pulsar schema type `EXTERNAL`. A topic cannot mix `EXTERNAL` schemas with native Pulsar schemas such as `JSON`, `AVRO`, or `PROTOBUF` on the same topic.

For example, if a topic already uses Pulsar's built-in Protobuf schema, creating a producer with External Protobuf Schema on the same topic fails with an incompatible schema error:

```
Incompatible schema: exists schema type PROTOBUF, new schema type EXTERNAL
```

<Warning>
  Plan your schema strategy before publishing to a topic. Once a topic uses External Protobuf Schema, all producers and consumers on that topic must use the same External Protobuf Schema type.
</Warning>

Schema compatibility modes for Protobuf in the Kafka Schema Registry are described in [Configurable compatibility modes](/cloud/governance/kafka-schemas/kafka-schema-registry#configurable-compatibility-modes). StreamNative Cloud supports a subset of the Kafka Schema Registry REST API. See the [REST API](/cloud/governance/kafka-schemas/kafka-schema-registry#rest-api) section for supported operations.

## Related resources

<CardGroup cols={2}>
  <Card title="External JSON Schema" icon="brackets-curly" href="/cloud/governance/kafka-schemas/external-json-schema">
    Use Kafka JSON Schema from Pulsar Java clients.
  </Card>

  <Card title="External Avro Schema" icon="file-code" href="/cloud/governance/kafka-schemas/external-avro-schema">
    Use Kafka Avro Schema from Pulsar Java clients.
  </Card>

  <Card title="Kafka Schema Registry" icon="database" href="/cloud/governance/kafka-schemas/kafka-schema-registry">
    Configure authentication, compatibility modes, and REST API access.
  </Card>

  <Card title="external-schemas repository" icon="github" href="https://github.com/streamnative/external-schemas">
    View source code, tests, and release notes for the kafka-schemas library.
  </Card>
</CardGroup>