Use External Avro Schema with Pulsar clients

External Avro Schema lets Pulsar Java clients produce and consume messages that use Kafka Avro 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 Avro Schema when you want to:

Use Pulsar clients with Kafka Avro Schema and Schema Registry compatibility checks.
Share Avro schemas between Kafka and Pulsar clients on the same topic.
Work with Avro SpecificRecord classes generated from .avsc schema files.

The kafka-schemas library provides a Pulsar Schema implementation backed by the Kafka Avro serializer. The same library also supports External JSON Schema and External Protobuf 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-avro-serializer</artifactId>
  <version>8.0.0</version>
</dependency>
<dependency>
  <groupId>org.apache.avro</groupId>
  <artifactId>avro</artifactId>
  <version>1.12.0</version>
</dependency>

The pulsar-client dependency provides the Pulsar Producer, Consumer, and PulsarClient APIs used in the examples below. The kafka-avro-serializer dependency is required at runtime because kafka-schemas declares it with provided scope. Declare avro explicitly so you control its version; kafka-schemas also pulls it in transitively.

Add the Kafka Maven repository

pulsar-client, kafka-schemas, and avro are available from Maven Central. You do not need to add a repository for those dependencies. kafka-avro-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-avro-serializer artifacts in an internal repository, configure that mirror instead of adding the public repository directly.

Define an Avro schema

External Avro Schema works with Avro SpecificRecord classes. Define a schema in an .avsc file and generate the Java class with the Avro Maven plugin.

Step 1: Create a schema file

Create src/main/avro/Player.avsc:

{
  "namespace": "com.example.avro",
  "type": "record",
  "name": "Player",
  "doc": "A player's profile information.",
  "fields": [
    {
      "name": "name",
      "type": "string",
      "doc": "The player's full name."
    },
    {
      "name": "number",
      "type": ["null", "int"],
      "default": null,
      "doc": "The player's number (optional)."
    },
    {
      "name": "favorite_color",
      "type": ["null", "string"],
      "default": null,
      "doc": "The player's favorite color (optional)."
    }
  ]
}

Step 2: Generate the SpecificRecord class

Add the Avro Maven plugin to your pom.xml:

<plugin>
  <groupId>org.apache.avro</groupId>
  <artifactId>avro-maven-plugin</artifactId>
  <version>1.12.0</version>
  <executions>
    <execution>
      <phase>generate-sources</phase>
      <goals>
        <goal>schema</goal>
      </goals>
      <configuration>
        <sourceDirectory>${project.basedir}/src/main/avro</sourceDirectory>
      </configuration>
    </execution>
  </executions>
</plugin>

Run mvn generate-sources to generate the Player class in the com.example.avro package.

Configure Schema Registry authentication

KafkaSchemaFactory accepts the same Schema Registry configuration properties as KafkaAvroSerializerConfig. 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(KafkaAvroSerializerConfig.SCHEMA_REGISTRY_URL_CONFIG, schemaRegistryUrl);
    configs.put(KafkaAvroSerializerConfig.BASIC_AUTH_CREDENTIALS_SOURCE, "USER_INFO");
    configs.put(
            KafkaAvroSerializerConfig.USER_INFO_CONFIG,
            String.format("%s:%s", "public", apiKey));
    return configs;
}

For additional serializer options, see the KafkaAvroSerializerConfig class in the kafka-avro-serializer dependency.

Produce and consume messages

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

import com.example.avro.Player;
import io.confluent.kafka.serializers.KafkaAvroSerializerConfig;
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 ExternalAvroSchemaExample {

    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/players";

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

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

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

        for (int i = 0; i < 10; i++) {
            Player player = new Player();
            player.setName("name-" + i);
            player.setNumber(i);
            player.setFavoriteColor("color-" + i);
            producer.send(player);
        }

        for (int i = 0; i < 10; i++) {
            Message<Player> message = consumer.receive();
            consumer.acknowledge(message);
            Player player = message.getValue();
            System.out.println("name=>" + player.getName()
                    + ", number=>" + player.getNumber()
                    + ", favoriteColor=>" + player.getFavoriteColor());
        }

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

    private static Map<String, Object> getSchemaRegistryConfigs(String schemaRegistryUrl, String apiKey) {
        Map<String, Object> configs = new HashMap<>();
        configs.put(KafkaAvroSerializerConfig.SCHEMA_REGISTRY_URL_CONFIG, schemaRegistryUrl);
        configs.put(KafkaAvroSerializerConfig.BASIC_AUTH_CREDENTIALS_SOURCE, "USER_INFO");
        configs.put(
                KafkaAvroSerializerConfig.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 Avro 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 Schema.AVRO(Player.class), creating a producer with External Avro Schema on the same topic fails with an incompatible schema error:

Incompatible schema: exists schema type AVRO, new schema type EXTERNAL

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

Schema compatibility modes for Avro 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 Protobuf Schema

Use Kafka Protobuf 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 an Avro schema

​Step 1: Create a schema file

​Step 2: Generate the SpecificRecord class

​Configure Schema Registry authentication

​Produce and consume messages

​Schema compatibility

​Related resources

External JSON Schema

External Protobuf Schema

Kafka Schema Registry

external-schemas repository

Prerequisites

Add the dependency

Add the Kafka Maven repository

Define an Avro schema

Step 1: Create a schema file

Step 2: Generate the SpecificRecord class

Configure Schema Registry authentication

Produce and consume messages

Schema compatibility

Related resources