Use External JSON Schema with Pulsar clients

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

Use Pulsar clients with Kafka JSON Schema and Schema Registry compatibility checks.
Share JSON schemas between Kafka and Pulsar clients on the same topic.
Build Key-Value messages where the key, value, or both use Kafka JSON Schema.

The kafka-schemas library (previously published as kafka-json-schema) provides a Pulsar Schema implementation backed by the Kafka JSON Schema serializer. The same library also supports External Avro 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-json-schema-serializer</artifactId>
  <version>8.0.0</version>
</dependency>

The pulsar-client dependency provides the Pulsar Producer, Consumer, and PulsarClient APIs used in the examples below. The kafka-json-schema-serializer dependency is required at runtime because kafka-schemas declares it with provided scope.

Add the Kafka Maven repository

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

Configure Schema Registry authentication

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

For additional serializer options, see the KafkaJsonSchemaSerializerConfig class in the kafka-json-schema-serializer dependency.

Produce and consume messages

The following example shows how to create a producer and consumer with External JSON Schema.

Step 1: Define your message class

Define a POJO for your message payload. Lombok annotations are optional.

public class User {
    private String name;
    private Integer age;

    public User() {}

    public User(String name, Integer age) {
        this.name = name;
        this.age = age;
    }

    public String getName() { return name; }
    public void setName(String name) { this.name = name; }
    public Integer getAge() { return age; }
    public void setAge(Integer age) { this.age = age; }
}

Step 2: Create a schema and connect to your cluster

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

import io.confluent.kafka.serializers.json.KafkaJsonSchemaSerializerConfig;
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 ExternalJsonSchemaExample {

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

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

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

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

        for (int i = 0; i < 10; i++) {
            producer.send(new User("name-" + i, 10 + i));
        }

        for (int i = 0; i < 10; i++) {
            Message<User> message = consumer.receive();
            consumer.acknowledge(message);
            System.out.println(message.getValue().getName());
        }

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

    private static Map<String, Object> getSchemaRegistryConfigs(String schemaRegistryUrl, String apiKey) {
        Map<String, Object> configs = new HashMap<>();
        configs.put(KafkaJsonSchemaSerializerConfig.SCHEMA_REGISTRY_URL_CONFIG, schemaRegistryUrl);
        configs.put(KafkaJsonSchemaSerializerConfig.BASIC_AUTH_CREDENTIALS_SOURCE, "USER_INFO");
        configs.put(
                KafkaJsonSchemaSerializerConfig.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.

Use Key-Value schemas

KafkaSchemaFactory also supports Key-Value messages. You can combine a native Pulsar schema for the key with External JSON Schema for the value, or use External JSON Schema for both key and value.

Native Pulsar key with External JSON Schema value

Use a native Pulsar Schema.STRING key and an External JSON Schema value:

import org.apache.pulsar.common.schema.KeyValue;
import org.apache.pulsar.common.schema.KeyValueEncodingType;

Schema<KeyValue<String, User>> schema = schemaFactory.kv(
        Schema.STRING,
        schemaFactory.json(User.class),
        KeyValueEncodingType.INLINE);

Producer<KeyValue<String, User>> producer = client.newProducer(schema).topic(topic).create();
producer.send(new KeyValue<>("user-1", new User("Alice", 30)));

External JSON Schema for both key and value

Use External JSON Schema for both the key and value:

public class UserKey {
    private Integer userId;
    private String name;
    // constructors and getters/setters omitted
}

Schema<KeyValue<UserKey, User>> schema = schemaFactory.kv(
        schemaFactory.json(UserKey.class),
        schemaFactory.json(User.class),
        KeyValueEncodingType.SEPARATED);

KeyValueEncodingType supports both INLINE and SEPARATED encoding, matching the behavior of Pulsar Key-Value schemas.

Schema compatibility

External JSON 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.JSON(User.class), creating a producer with External JSON Schema on the same topic fails with an incompatible schema error:

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

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

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

Use Kafka Avro 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

​Configure Schema Registry authentication

​Produce and consume messages

​Step 1: Define your message class

​Step 2: Create a schema and connect to your cluster

​Use Key-Value schemas

​Native Pulsar key with External JSON Schema value

​External JSON Schema for both key and value

​Schema compatibility

​Related resources

External Avro Schema

External Protobuf Schema

Kafka Schema Registry

external-schemas repository

Prerequisites

Add the dependency

Add the Kafka Maven repository

Configure Schema Registry authentication

Produce and consume messages

Step 1: Define your message class

Step 2: Create a schema and connect to your cluster

Use Key-Value schemas

Native Pulsar key with External JSON Schema value

External JSON Schema for both key and value

Schema compatibility

Related resources