Kafka — Schema Registry

Configure a Kafka consumer to decode Confluent Schema Registry-framed messages, and a publisher to produce them. On each consumed message the consumer strips the wire frame (magic byte 0x00 + big-endian schema id), resolves the schema from the registry (cached in memory), validates the schema subject against the accepted set, then delegates to the topic's Codec for the actual payload decode. On the produce side, an opt-in publisher wraps each encoded payload in the same frame — see Producing SR-framed messages.

Works with the Confluent Schema Registry and with Redpanda's built-in Schema Registry — both expose the same REST API and wire format.

Prerequisites

• Cargo features: kafka-schema-registry (implies kafka)
• A running Confluent-compatible Schema Registry (Confluent, Redpanda, or compatible)
• Topics must use JsonCodec or ProtobufCodec; other codecs are not supported

Install

cargo add shove --features kafka-schema-registry

Build the registry client

use std::time::Duration;
use shove::schema_registry::{SchemaRegistry, SchemaRegistryAuth};

let registry = SchemaRegistry::builder("https://schema-registry:8081")
    .auth(SchemaRegistryAuth::Basic {
        user: "sr-user".into(),
        pass: "sr-pass".into(),
    })
    .timeout(Duration::from_secs(3))
    .max_retries(2)
    .negative_cache_ttl(Duration::from_secs(60))
    .build();

build() returns Arc<SchemaRegistry>. Clone the Arc to share the same schema cache across consumers or a consumer group.

Auth options:

Credentials require TLS

A credential configured against a base URL that is not https:// makes build() panic — the secret would otherwise be sent in cleartext on every schema fetch, readable by anything on the path. An unauthenticated http:// registry keeps working unchanged.

Three configurations count as a credential, because all three put a reusable secret on the wire:

The panic message names which of them it found, never their values.

The base URL is parsed before it is judged rather than pattern-matched as a string. url treats a missing or repeated // after a special scheme as a recoverable syntax violation, so http:registry:8081, http:/registry:8081 and http:///registry:8081 all reach the same origin as http://registry:8081 — and are all refused. A base URL that does not parse cannot be shown to be encrypted, so it is treated as plaintext.

A credential-bearing client additionally refuses to follow redirects, and surfaces a 3xx as a transport error. Following one would replay a custom secret header to whatever host and scheme the Location names; the HTTP client only strips Authorization, Cookie, Proxy-Authorization and WWW-Authenticate across an origin change, and a header such as CF-Access-Client-Secret is on none of those lists. An unauthenticated client follows redirects as before.

For a registry genuinely unreachable from an untrusted network, such as a local development stack, opt in explicitly:

use shove::schema_registry::{SchemaRegistry, SchemaRegistryAuth};

let registry = SchemaRegistry::builder("http://localhost:8081")
    .auth(SchemaRegistryAuth::Bearer("dev-token".into()))
    .allow_plaintext_credentials()
    .build();

Custom headers

.header(name, value) attaches a static header to every registry request. Its value is treated as a secret: the builder cannot tell which vendor-specific header name holds a credential, and this method exists mainly to carry them (Cloudflare Access, API gateways). So a .header(...) call makes the registry credential-bearing — TLS required, no redirects, and the value is marked sensitive so it prints as Sensitive rather than in clear.

For a header that genuinely carries no secret, use .non_secret_header(name, value). It is identical on the wire; the difference is the assertion you are making:

use shove::schema_registry::SchemaRegistry;

let registry = SchemaRegistry::builder("https://schema-registry:8081")
    // A credential: requires TLS, redacted in diagnostics.
    .header("CF-Access-Client-Secret", "…")
    // Not a credential: no TLS requirement of its own.
    .non_secret_header("X-Client-Build", "2026.08.30")
    .build();

When a benign header is the only thing tripping the guard, move it to non_secret_header rather than reaching for allow_plaintext_credentials(): that opt-in is not scoped to a single header, so using it to quiet an Accept header would also permit a real bearer token in cleartext.

non_secret_header is an assertion about the value, not a way to silence the guard. Using it on a header that does carry a secret disables both the TLS requirement and the redirect protection for that secret, with nothing to warn you. Authorization, Proxy-Authorization and Cookie are rejected outright, since no configuration makes that assertion true for them — but a vendor-specific name like CF-Access-Client-Secret cannot be recognised, and only you know what is in it. When in doubt, use .header(...).

Per-consumer setup

use std::sync::Arc;
use shove::schema_registry::{SchemaEnforcement, SchemaRegistry};
use shove::consumer::ConsumerOptions;
use shove::markers::Kafka;

let registry = SchemaRegistry::builder("http://schema-registry:8081").build();

let opts = ConsumerOptions::<Kafka>::new()
    .with_schema_registry(Arc::clone(&registry))
    .with_schema_enforcement(SchemaEnforcement::Enforce)
    .accept_schema_subjects(["orders-value"]);

Per-consumer-group setup

Attaching the registry on KafkaConsumerGroupConfig shares one Arc — and therefore one schema cache — across the whole autoscaling group:

use std::sync::Arc;
use shove::kafka::KafkaConsumerGroupConfig;
use shove::schema_registry::{SchemaEnforcement, SchemaRegistry};

let registry = SchemaRegistry::builder("http://schema-registry:8081").build();

let cfg = KafkaConsumerGroupConfig::new(1..=8)
    .with_schema_registry(Arc::clone(&registry))
    .with_schema_enforcement(SchemaEnforcement::Enforce)
    .accept_schema_subjects(["orders-value"]);

Producing SR-framed messages

The same registry client drives producer-side encoding. Attach it to a KafkaPublisherConfig and get the publisher via Broker::publisher_with — every publish then emits a Confluent-framed payload, wire-compatible with any standard Confluent consumer (including ClickPipes and shove's own decode path above).

use std::sync::Arc;
use shove::kafka::KafkaPublisherConfig;
use shove::schema_registry::SchemaRegistry;

let registry = SchemaRegistry::builder("http://schema-registry:8081").build();

// `broker` is a `Broker<Kafka>`.
let publisher = broker
    .publisher_with(KafkaPublisherConfig::new().with_schema_registry(Arc::clone(&registry)))
    .await?;

// Emits `0x00` + big-endian schema id + encoded payload.
publisher.publish::<OrdersTopic>(&order).await?;

The publisher looks up the latest schema id for the subject (GET /subjects/{subject}/versions/latest, cached per subject) — it carries no schema text, so the subject must already be registered. The subject defaults to "{topic}-value"; override it with KafkaPublisherConfig::with_subject("..."). Framing applies only to JsonCodec / ProtobufCodec topics (Protobuf uses message index [0]); a plain Broker::publisher() does no framing.

Enforcement modes

SchemaEnforcement controls what happens when a message's registered subject is not in the accepted set:

• Enforce (default) — the message is routed to the DLQ with death reason schema_validation_failed. Use this in production where a subject mismatch is a producer misconfiguration.
• Permissive — the mismatch is logged and counted; the message is still decoded. Use this during migration windows or when producers share a topic with different subject conventions.

Accepted subjects

accept_schema_subjects([...]) pins the set of Confluent schema subjects the consumer will accept. When not set, the default follows the Confluent TopicNameStrategy: "{queue}-value". For a topic named orders that resolves to orders-value.

DLQ consumer caveat

What to try next

• Change SchemaEnforcement::Enforce to SchemaEnforcement::Permissive and verify that subject-mismatched messages reach your handler rather than the DLQ.
• Share one Arc<SchemaRegistry> across two different ConsumerGroupConfig registrations to confirm they hit the same cache.
• Point SchemaRegistry::builder at a Redpanda broker URL to confirm Redpanda Schema Registry compatibility.
• See the Kafka backend overview for the full configuration reference.