Schema enforcement
This product is in beta. For more information about using Bufstream in production, contact us.
Data quality enforcement
Rather than trusting producers to send valid data, Bufstream can reject messages that don't match the topic's schema. This guarantees that consumers will always receive well-formed data, eliminating a large class of data outages. This feature works with binary-encoded Protobuf messages and the Buf Schema Registry (or any other Protobuf registry that supports Confluent's REST API). To associate Protobuf schemas with a Kafka topic in the Buf Schema Registry, follow the documentation for integrating the BSR with Kafka.
Self-describing data
It's often helpful to make Kafka messages self-describing, so that tools and frameworks can unmarshal, manipulate, and display them. The most common approach to making messages self-describing is to prefix the serialized message with a few extra bytes (commonly called an "envelope"). The prefix encodes the ID of the message's schema, which can then be retrieved from a schema registry. Because Confluent introduced and popularized this format, it's commonly called the Confluent wire format. Much of the Kafka ecosystem supports this format, including most client libraries, Kafka Connect, AKHQ, kSQLdb, and Snowflake's Kafka Connector.
Enabling schema enforcement
To reject data that doesn't match the topic's schema, add the following to your Helm values file:
data_enforcement:
schema_registries:
- name: bsr
confluent:
url: https://my-schema-registry.com/registry-id
basic_auth:
username:
string: your-user
password:
string: your-password
produce:
- topics: { equal: my-topic }
schema_registry: my-registry
values:
coerce: true
on_parse_error: REJECT_BATCH
With this configuration, consumers always receive well-formed, self-describing data. If the producer sends un-enveloped messages, Bufstream automatically envelopes them using the latest revision of the topic's schema. Bufstream then parses the message using the schema from the envelope. If the message doesn't match the schema, Bufstream rejects it and the data producer sees an error.
Advanced configurations
Automatic enveloping without parsing
To enable automatic enveloping without verifying that the data matches the schema, change the produce filter above to the following:
produce:
- topics: { all: true }
schema_registry: my-registry
values:
coerce: true
skip_parse: true
Enforce producer enveloping without parsing
To require that data producers send enveloped messages without verifying that the data matches the schema, change the produce filter to the following:
produce:
- topics: { all: true }
schema_registry: my-registry
values:
coerce: false
on_no_schema: REJECT_BATCH
Automatically enveloping old messages
If a topic contains un-enveloped data, Bufstream can envelope the existing data on the fly as it's read. To automatically envelope the data and verify that it matches the schema, hiding messages with invalid data from the consumer, add the following to your Helm values file:
data_enforcement:
schema_registries:
- name: bsr
confluent:
url: https://my-schema-registry.com/registry-id
basic_auth:
username:
string: your-user
password:
string: your-password
fetch:
- topics: { all: true }
schema_registry: my-registry
values:
coerce: true
on_parse_error: FILTER_RECORD
To instead allow the consumer to see enveloped but invalid data, change FILTER_RECORD
to PASS_THROUGH
.
Topic-specific policies
For finer-grained control, Bufstream supports topic-specific policies. Only the first policy that matches a given produce or fetch request takes effect. This Helm values snippet shows how to configure a policy for a specific topic:
produce:
- topics: { equal: "topic" }
schema_registry: my-registry
values:
coerce: true
on_parse_error: REJECT_BATCH
Using multiple registries
There may be cases where you have more than one registry in place — for example, a development and production registry. Bufstream allows you to configure each registry independently and choose which registry to use with each policy. As an example, this Helm values snippet configures two registries and uses each in one policy:
produce:
# pass through any errors from the development registry for a specific topic ("dev-topic").
- topics: { equal: "dev-topic" }
schema_registry: dev-registry
values:
coerce: true
on_parse_error: PASS_THROUGH
# reject messages that error for all topics in the production registry
- topics: { all: true }
schema registry: prod-registry
values:
coerce: true
on_parse_error: REJECT_BATCH