Schema Registry

SchemaRegistry is the Confluent Schema Registry client: it sends nothing until the first lookup, and its clones share one cache.

SchemaRegistry::new records the base URL, so the app starts with no registry reachable. basic_auth and bearer_token set the credentials of every request, and HTTPS goes through rustls. request_timeout bounds each request, ten seconds by default, so a registry that accepts the connection and goes silent fails the lookup with KafkaError::SchemaRegistryTimeout. The client is configured before it is handed out, because clones keep the settings they were made from.

Every clone reads and writes one cache, so a schema or a subject is fetched once per process. schema_by_id answers from the cache or asks the registry once; cached_schema and cached_subject read the cache alone. A RegisteredSchema carries the id, the SchemaType (Avro, Protobuf or JSON) and the definition text.

A registry-backed payload carries the Confluent envelope: a zero byte, the schema id, then the datum. A broker meets it in one of two ways. The codecs, AvroCodec and SchemaFramed, hold the format where a serializer belongs and take their schemas through a SchemaPrefetch. The transcode converts at the broker's edges and leaves the handlers on plain JSON. The example holds the client as application state, and a handler asks it which schema a record was written with.

Example

Macros
Manual

examples/kafka_registry_client.rs
/// A record as it arrived, envelope included.
#[derive(Deserialized)]
struct Framed<'a>(&'a [u8]);

#[derive(FromRef)]
struct Audit {
    registry: SchemaRegistry,
}

#[subscriber("orders")]
async fn audit(record: &Framed<'_>, State(registry): State<SchemaRegistry>) -> HandlerOutcome {
    let Some((id, _)) = parse_envelope(record.0) else {
        tracing::info!(topic = "orders", "a record without a schema");
        return HandlerOutcome::ack();
    };
    // The first lookup of an id reaches the registry; every later one reads the cache.
    match registry.schema_by_id(id).await {
        Ok(schema) => {
            tracing::info!(schema_id = id, schema_type = ?schema.schema_type(), "written with");
            HandlerOutcome::ack()
        }
        Err(error) => {
            tracing::warn!(topic = "orders", schema_id = id, %error, "schema lookup failed");
            HandlerOutcome::retry()
        }
    }
}

fn registry(url: &str, password: &str) -> SchemaRegistry {
    // Records the address and the settings; no request is sent until the first lookup.
    SchemaRegistry::new(url)
        .basic_auth("audit-svc", password)
        .request_timeout(Duration::from_secs(2))
}

fn app(registry: SchemaRegistry) -> impl App {
    let broker = KafkaBroker::new(["localhost:9092"]).default_group("audit-svc");
    RustStream::new(AppInfo::new("audit", "0.1.0"))
        .on_startup(async move |()| Ok::<_, Infallible>(Audit { registry }))
        .with_broker(broker, |b| {
            b.include(audit);
        })
}

#[ruststream::app]
fn service() -> impl App {
    let password = std::env::var("REGISTRY_PASSWORD").unwrap_or_default();
    app(registry("http://localhost:8081", &password))
}

examples/manual/kafka_registry_client.rs
/// A record as it arrived, envelope included.
struct Framed<'a>(&'a [u8]);

// `#[derive(Deserialized)]` over a borrowed byte slice: the view is the payload itself.
impl Deserialized for Framed<'_> {
    type Output<'a> = Framed<'a>;
    type Error = Infallible;

    fn from_payload(payload: &[u8]) -> Result<Framed<'_>, Infallible> {
        Ok(Framed(payload))
    }
}

impl Input for Framed<'_> {
    type Axis = SoloDeserialized<Framed<'static>>;
}

/// The application state the handler reads.
struct Audit {
    registry: SchemaRegistry,
}

/// The definition `#[subscriber("orders")]` would have minted; the state axis names `Audit`.
struct AuditRecord;

impl Handle<Framed<'_>, (), (), (), Audit> for AuditRecord {
    async fn handle(
        &self,
        record: &Framed<'_>,
        _outs: &(),
        ctx: &mut Context<'_, (), Audit>,
    ) -> Result<(), HandlerOutcome> {
        let Some((id, _)) = parse_envelope(record.0) else {
            tracing::info!(topic = "orders", "a record without a schema");
            return Ok(());
        };
        // The first lookup of an id reaches the registry; every later one reads the cache.
        match ctx.state().registry.schema_by_id(id).await {
            Ok(schema) => {
                tracing::info!(schema_id = id, schema_type = ?schema.schema_type(), "written with");
                Ok(())
            }
            Err(error) => {
                tracing::warn!(topic = "orders", schema_id = id, %error, "schema lookup failed");
                Err(HandlerOutcome::retry())
            }
        }
    }
}

fn registry(url: &str, password: &str) -> SchemaRegistry {
    // Records the address and the settings; no request is sent until the first lookup.
    SchemaRegistry::new(url)
        .basic_auth("audit-svc", password)
        .request_timeout(Duration::from_secs(2))
}

fn app(registry: SchemaRegistry) -> impl App {
    let broker = KafkaBroker::new(["localhost:9092"]).default_group("audit-svc");
    RustStream::new(AppInfo::new("audit", "0.1.0"))
        .on_startup(async move |()| Ok::<_, Infallible>(Audit { registry }))
        .with_broker(broker, |b| {
            b.include(subscriber("orders", AuditRecord).build());
        })
}

// What `#[ruststream::app]` writes: the command line on a multi-threaded Tokio runtime.
fn main() -> ExitCode {
    let password = std::env::var("REGISTRY_PASSWORD").unwrap_or_default();
    run_main(|| app(registry("http://localhost:8081", &password)))
}

Example​

Example