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

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

External Avro Schema lets Pulsar Java clients produce and consume messages that use [Kafka Avro 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 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`](https://github.com/streamnative/external-schemas) library provides a Pulsar `Schema` implementation backed by the Kafka Avro serializer. The same library also supports [External JSON Schema](/cloud/governance/kafka-schemas/external-json-schema) and [External Protobuf Schema](/cloud/governance/kafka-schemas/external-protobuf-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-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](https://repo1.maven.org/maven2/). 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`:

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

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

## 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`:

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

```xml theme={null}
<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`](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(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.

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

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

Schema compatibility modes for Avro 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 Protobuf Schema" icon="code" href="/cloud/governance/kafka-schemas/external-protobuf-schema">
    Use Kafka Protobuf 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>