Schema Registry

Applies to VAST Cluster 5.5.

Confluent Schema Registry stores Avro, Protobuf and JSON Schema schemas, and its serializers encode event values against them. With VAST Event Broker you can use the serializers in your producers and consumers as you would with any Kafka cluster, because they run in the client: the broker only stores the bytes they produce. The registry itself is a separate service that keeps its schemas in a Kafka topic, and that part needs care. This page covers both, and what registry-encoded values mean for blob expansion and for reading topics as tables.

The examples use Confluent Schema Registry 8.0.0 and confluent-kafka 2.8.0 serializers: Avro and JSON Schema producers and consumers against Event Broker topics, the registry's storage on Event Broker (fails, as below) and on an Apache Kafka cluster (works), and blob expansion of registry-encoded values. Protobuf, 7.x registry releases and secured connections aren't covered.

Serializers: nothing changes on the broker

The Avro, Protobuf and JSON Schema serializers run in your producer. Each one looks up or registers the schema with the registry over HTTP, then encodes the value and passes plain bytes to the Kafka client. The deserializers do the reverse in your consumer. The broker never talks to the registry and doesn't check the bytes: it stores them in the topic's value (or key) column like any other payload.

So a producer that uses a serializer needs the same settings as any producer, and nothing more: idempotence off and no compression. See Producing. For example, with the confluent-kafka Python client:

import os

from confluent_kafka import Producer
from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.avro import AvroSerializer
from confluent_kafka.serialization import MessageField, SerializationContext

TOPIC = "docs-schema-registry-orders"
SCHEMA = """
{"type": "record", "name": "Order",
 "fields": [{"name": "order_id", "type": "long"}, {"name": "status", "type": "string"}]}
"""

registry = SchemaRegistryClient({"url": os.environ["SCHEMA_REGISTRY_URL"]})
serialize = AvroSerializer(registry, SCHEMA)

producer = Producer({
    "bootstrap.servers": os.environ["KAFKA_BOOTSTRAP"],
    "enable.idempotence": False,
    "compression.type": "none",
})
order = {"order_id": 1, "status": "new"}
producer.produce(TOPIC, key=b"order-1",
                 value=serialize(order, SerializationContext(TOPIC, MessageField.VALUE)))
producer.flush()

The registry doesn't have to use the same Kafka cluster as your data. That's what makes the next section workable.

The registry's own storage

The registry keeps every schema in one Kafka topic, set by kafkastore.topic (default _schemas), on the cluster set by kafkastore.bootstrap.servers. The topic must have a single partition and be compacted. The registry creates it on first start if it doesn't exist, and on every start it checks the partition count and the cleanup policy. The same cluster is used to elect the leader among registry instances.

Event Broker supports compacted topics and ignores the replication factor, so the topic itself should be fine. The problem is the registry's producer:

Recommendation: keep the registry's storage on a small Apache Kafka cluster, and your data topics on Event Broker. Your producers and consumers reach the registry only through its URL (schema.registry.url, or url in the Python client), so they don't care where it stores schemas.

# schema-registry.properties: storage on a separate Apache Kafka cluster
listeners = http://0.0.0.0:8081
kafkastore.bootstrap.servers = PLAINTEXT://kafka-1:9092
kafkastore.topic = _schemas

Blob expansion and registry-encoded values

Blob expansion parses each event's value as JSON. See Blob expansion. A value written by a Schema Registry serializer isn't JSON, even with JSON Schema: by default each serializer puts a five-byte prefix before the data, a version byte (0) and the four-byte schema ID, and the Protobuf serializer adds message indexes after it (Confluent wire format).

So if you attach a blob expansion to a topic of registry-encoded values, every event fails to parse. Each produce still succeeds, and each target row has every expanded field null and blob_expansion_error_output = 1.

For a topic you want to expand, produce plain JSON values, for example json.dumps(order).encode(), without a schema serializer. If you also need schema checks, run them in the producer before you send.

For example, a JSON Schema-encoded order expands to a row with every field null and blob_expansion_error_output = 1, while a plain JSON order produced to the same topic expands normally.

Querying registry-encoded topics

Every topic is a VAST DataBase table, and you can read it with the SDK, the Query Engine, Trino or Spark. See Query topics as tables. For a registry-encoded topic, value holds the encoded bytes, prefix included. The cluster doesn't decode them, and SQL can't filter on fields inside them. Read the bytes and decode them in your client with the deserializer for the format:

import os

import vastdb
from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.avro import AvroDeserializer
from confluent_kafka.serialization import MessageField, SerializationContext

TOPIC = "docs-schema-registry-orders"
registry = SchemaRegistryClient({"url": os.environ["SCHEMA_REGISTRY_URL"]})
deserialize = AvroDeserializer(registry)   # looks up each value's writer schema by its ID

session = vastdb.connect()
with session.transaction() as tx:
    topic = tx.bucket(os.environ["EVENTS_BUCKET"]).schema("kafka_topics").table(TOPIC)
    rows = topic.select(columns=["key", "value"]).read_all()

ctx = SerializationContext(TOPIC, MessageField.VALUE)
for row in rows.to_pylist():
    print(row["key"], deserialize(row["value"], ctx))

Filters on the cluster can use the other columns, such as key and the timestamps, but not fields inside value. If you need to filter or aggregate on fields, produce plain JSON and use blob expansion, or decode the values and write them to a table of your own.

For example, a stored value starts with 00 00 00 00 01 (Avro, schema ID 1) or 00 00 00 00 02 (JSON Schema, schema ID 2), followed by the encoded record.

What won't work

• Running a current registry release with its storage on Event Broker: its storage producer is idempotent, and you can't turn that off.
• Blob expansion of registry-encoded values: they aren't plain JSON.
• SQL on fields inside encoded values: the cluster sees bytes.

See also

• Integrations: the checklist for any Kafka tool.
• Producing: required producer settings.
• Blob expansion and Query topics as tables
• Kafka compatibility
• Confluent Schema Registry configuration reference and Formats, serializers and deserializers