Use External Protobuf Schema with Pulsar clients

External Protobuf Schema lets Pulsar Java clients produce and consume messages that use Kafka Protobuf Schema and the 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 library provides a Pulsar Schema implementation backed by the Kafka Protobuf serializer. The same library also supports External JSON Schema and External Avro Schema.

Prerequisites

A StreamNative Pulsar cluster for message production and consumption.
The 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 role to register schemas and the schema-reader role to read schemas. See Schema Registry RBAC roles.
Java 17 or higher.
Pulsar Java client 4.1.0 or higher.

Add the dependency

Add the following Maven dependencies to your project:

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

<repositories>
  <repository>
    <id>kafka</id>
    <url>https://packages.confluent.io/maven/</url>
  </repository>
</repositories>

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

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:

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:

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:

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

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.

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

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.

Schema compatibility modes for Protobuf in the Kafka Schema Registry are described in Configurable compatibility modes. StreamNative Cloud supports a subset of the Kafka Schema Registry REST API. See the REST API section for supported operations.

External JSON Schema

Use Kafka JSON Schema from Pulsar Java clients.

External Avro Schema

Use Kafka Avro Schema from Pulsar Java clients.

Kafka Schema Registry

Configure authentication, compatibility modes, and REST API access.

external-schemas repository

View source code, tests, and release notes for the kafka-schemas library.

​Prerequisites

​Add the dependency

​Add the Kafka Maven repository

​Define a Protobuf schema

​Step 1: Create Protobuf definition files

​Step 2: Generate the Protobuf classes

​Configure Schema Registry authentication

​Produce and consume messages

​Schema compatibility

​Related resources

External JSON Schema

External Avro Schema

Kafka Schema Registry

external-schemas repository

Prerequisites

Add the dependency

Add the Kafka Maven repository

Define a Protobuf schema

Step 1: Create Protobuf definition files

Step 2: Generate the Protobuf classes

Configure Schema Registry authentication

Produce and consume messages

Schema compatibility

Related resources