Crate crabka_schema_registry 

Confluent Schema Registry-compatible REST service for Crabka.

Standalone binary; a Kafka client of a Crabka broker. State lives in the _schemas compacted topic.

Runtime configuration

use crabka_schema_registry::config::{RegistryConfig, SecurityConfig};

let config = RegistryConfig {
    bootstrap: "localhost:9092".into(),
    schemas_topic: "_schemas".into(),
    schemas_topic_rf: 3,
    client_id: "schema-registry-1".into(),
    advertised_url: "http://schema-registry-1:8081".into(),
    group_id: "schema-registry".into(),
    leader_eligibility: true,
    security: SecurityConfig::default(),
};

assert_eq!(config.schemas_topic, "_schemas");

Compatibility checks

use crabka_schema_registry::format::{self, SchemaType};

let prior = r#"{"type":"record","name":"Order","fields":[{"name":"id","type":"string"}]}"#;
let next = r#"{"type":"record","name":"Order","fields":[{"name":"id","type":"string"},{"name":"total","type":["null","double"],"default":null}]}"#;

assert!(format::check(SchemaType::Avro, next, prior, &[], &[]).is_ok());

Modules

auth

Authentication middleware: resolve a crabka_security::Principal from each request (mTLS → Bearer → Basic → Anonymous) into request extensions; 401 on a bad credential or a missing one when require_auth. Reuses crabka_security validators; only BasicAuthStore is local. Models grpc-gateway/src/authz/auth_layer.rs.

authz

Topic-ACL authorization for the registry REST surface.

cli

Clap-free CLI → SecurityConfig assembly.

compat

Compatibility engine: resolve the effective level, pick the version set, and run the per-format directional check. Format-agnostic (delegates to format::check); knows nothing about Avro/Protobuf/JSON internals.

config

Runtime configuration for the registry service.

election

Schema Registry primary election (cp-exact "sr" Kafka group). A node joins the group; the leader selects the primary and broadcasts it; every node publishes its PrimaryState for the forwarding middleware.

error

Confluent-compatible error model: numeric error_code + HTTP status, serialised as {"error_code":N,"message":"..."} with the vendor content-type. Serdes branch on error_code, so the numbers are exact.

format

Schema formats: parse, canonical storage form, and directional compatibility checks. Canonical form is the global-id deduplication key.

kafkastore

Kafka-backed schema store — reads/writes from the _schemas compacted topic.

rest

HTTP surface: AppState + the merged Confluent route table.

store

In-memory authoritative registry state, rebuilt by replaying _schemas. Pure data structure: no I/O. The KafkaStore wraps it behind a lock and the write-serialisation gate (see kafkastore/mod.rs). Cloneable so the write path can decide id/version on a throwaway copy (the reader is the sole mutator of the live instance).