Schema Registry: Avro и JSON Schema

Это руководство читает и публикует записи Avro и JSON Schema в конверте Confluent и при запуске сверяет каждую схему со Schema Registry.

Запись, которая опирается на реестр, начинается с конверта Confluent: нулевого байта и четырёхбайтового идентификатора схемы, которой она записана. Сами схемы хранятся в реестре. В RustStream схема принадлежит кодеку, поэтому обработчики принимают и возвращают обычные структуры serde.

Cargo.toml

[dependencies]
ruststream-rdkafka = { version = "0.7", features = ["avro"] }
apache-avro = { version = "0.22", features = ["derive"] }

Фича avro добавляет кодек Avro поверх schema-registry, а та приносит клиент реестра и конверт. apache-avro даёт derive AvroSchema. Берите ту версию, от которой зависит крейт.

Модели

Макросы

src/main.rs

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, AvroSchema)]
struct Invoice {
    id: i64,
    amount_cents: i64,
    // Added in the second version of the schema; older producers do not write it.
    #[avro(default = r#""EUR""#)]
    currency: String,
}

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, AvroSchema, Outgoing)]
#[outgoing(name = "receipts")]
struct Receipt {
    invoice: i64,
    amount_cents: i64,
    currency: String,
}

Вручную

src/main.rs

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, AvroSchema, JsonSchema)]
struct Invoice {
    id: i64,
    amount_cents: i64,
    // Added in the second version of the schema; older producers do not write it.
    #[avro(default = r#""EUR""#)]
    currency: String,
}

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, AvroSchema, JsonSchema)]
struct Receipt {
    invoice: i64,
    amount_cents: i64,
    currency: String,
}

// What `#[derive(Outgoing)]` with `#[outgoing(name = "receipts")]` writes.
impl OutgoingDestination for Receipt {
    type Form = FixedName;

    const DESTINATION: &'static str = "receipts";
}

impl MessageHeaders for Receipt {
    type Contract = NoHeaders;
}

impl MessageInfo for Receipt {
    const NAME: &'static str = "Receipt";
    const DESCRIPTION: Option<&'static str> = None;
}

AvroSchema выводит схему Avro каждой структуры. Поле currency появилось во второй версии схемы счёта. #[avro(default)] - значение, которое получает запись, записанная без этого поля.

Обработчик

Макросы

src/main.rs

#[subscriber("invoices", reply)]
async fn issue_receipt(invoice: &Invoice) -> Receipt {
    tracing::info!(invoice.id, invoice.amount_cents, "receipt issued");
    Receipt {
        invoice: invoice.id,
        amount_cents: invoice.amount_cents,
        currency: invoice.currency.clone(),
    }
}

Вручную

src/main.rs

struct IssueReceipt;

impl Handle<Invoice, Receipt> for IssueReceipt {
    fn handle(
        &self,
        invoice: &Invoice,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<Receipt, HandlerOutcome>> {
        tracing::info!(invoice.id, invoice.amount_cents, "receipt issued");
        ready(Ok(Receipt {
            invoice: invoice.id,
            amount_cents: invoice.amount_cents,
            currency: invoice.currency.clone(),
        }))
    }
}

issue_receipt читает счёт и отвечает квитанцией в тему receipts. Ничего в нём не знает ни об Avro, ни о конверте.

Реестр, предвыборка и кодек

Макросы

src/main.rs

fn service(broker: KafkaBroker, registry: SchemaRegistry) -> impl App {
    let prefetch = SchemaPrefetch::new(registry);
    let avro = AvroCodec::registry(&prefetch)
        .register::<Receipt>("receipts-value")
        .resolve_onto(Invoice::get_schema())
        .expect("a registry codec resolves its reader schema on the first delivery");
    let json = SchemaFramed::new(&prefetch, JsonCodec);

    RustStream::new(AppInfo::new("billing", "0.1.0")).with_broker_codec(
        broker.schema_prefetch(prefetch),
        avro.clone(),
        |b| {
            b.include(issue_receipt)
                .out_reply(KafkaPublish::default())
                .codec(avro);
            b.include_router(Router::new().with_codec(json).include(open_dispute));
        },
    )
}

#[ruststream::app]
fn app() -> impl App {
    service(
        KafkaBroker::new(["localhost:9092"]).default_group("billing-svc"),
        SchemaRegistry::new("http://localhost:8081"),
    )
}

Вручную

src/main.rs

fn service(broker: KafkaBroker, registry: SchemaRegistry) -> impl App {
    let prefetch = SchemaPrefetch::new(registry);
    let avro = AvroCodec::registry(&prefetch)
        .register::<Receipt>("receipts-value")
        .resolve_onto(Invoice::get_schema())
        .expect("a registry codec resolves its reader schema on the first delivery");
    let json = SchemaFramed::new(&prefetch, JsonCodec);

    RustStream::new(AppInfo::new("billing", "0.1.0")).with_broker_codec(
        broker.schema_prefetch(prefetch),
        avro.clone(),
        |b| {
            b.include(subscriber("invoices", IssueReceipt).reply().build())
                .out_reply(KafkaPublish::default())
                .codec(avro);
            b.include_router(
                Router::new()
                    .with_codec(json)
                    .include(subscriber("disputes", OpenDispute).build()),
            );
        },
    )
}

fn app() -> impl App {
    service(
        KafkaBroker::new(["localhost:9092"]).default_group("billing-svc"),
        SchemaRegistry::new("http://localhost:8081"),
    )
}

fn main() -> ExitCode {
    run_main(app)
}

SchemaRegistry::new запоминает URL реестра и пока ничего не отправляет. Все его клоны делят один кеш, поэтому схема загружается один раз на процесс. Сетевую работу выполняет SchemaPrefetch, а schema_prefetch подключает его к брокеру. Когда брокер подключается, предвыборка разрешает каждый субъект, под которым публикуют кодеки. Когда приходит запись, предвыборка ещё до работы кодека загружает схему, которую называет конверт: ту, по которой запись записана. Поэтому сам кодек никогда не ждёт сети.

AvroCodec::registry строит кодек Avro из предвыборки. register называет субъект, под которым оформляются значения типа: квитанции уходят под receipts-value. Для декодирования регистрация не нужна, потому что каждая запись сама называет схему, по которой её записали.

with_broker_codec декодирует все обработчики области кодеком Avro. Ответ кодируется кодеком, названным в его точке монтирования, а если кодек не назван, то JSON-кодеком по умолчанию. Поэтому out_reply(KafkaPublish::default()).codec(avro) называет для квитанций кодек Avro.

Чтение записей по старой схеме

resolve_onto велит Avro приводить каждую запись к схеме чтения, здесь Invoice::get_schema(). У записи, записанной по первой версии схемы, нет currency. Приведение заполняет поле значением по умолчанию, и обработчик читает "EUR". Схема чтения действует на каждую запись, которую декодирует её кодек, поэтому один такой кодек обслуживает один тип. Второй тип со своей схемой чтения получает собственный кодек в собственном роутере.

Регистрация субъектов и отсутствующий субъект

Сервис не регистрирует схем: субъект регистрирует тот, кому принадлежит тема, например на шаге развёртывания. register_avro регистрирует схему Avro типа под субъектом, register_json - его JSON Schema, а register - схему, заданную текстом.

Если субъекта, под которым публикует кодек, в реестре нет, брокер не подключается, и ошибка называет субъект. MissingSubject выбирает, что происходит вместо этого:

Политика	Что она делает
Refuse (по умолчанию)	сервис не запускается
AutoRegister	регистрирует под субъектом собственную схему типа и пишет предупреждение в журнал
PublishUnframed	публикует голые данные без конверта и пишет предупреждение в журнал

Задайте её через on_missing_subject до того, как из предвыборки будут построены кодеки.

Совместимость при запуске

При подключении схема каждого зарегистрированного типа проверяется так, как проверялась бы при регистрации. Уровень совместимости субъекта решает, с какими версиями она сверяется: с последней или, при уровне _TRANSITIVE, со всеми. Модель, которая разошлась со своим субъектом, останавливает сервис, а описание расхождения от реестра попадает в ошибку. check_compatibility(false) выключает проверку для субъекта с уровнем NONE.

JSON Schema в конверте

Макросы

src/main.rs

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
struct Dispute {
    invoice: i64,
    reason: String,
}

#[subscriber("disputes")]
async fn open_dispute(dispute: &Dispute) -> HandlerOutcome {
    if dispute.reason.is_empty() {
        tracing::warn!(dispute.invoice, "dispute without a reason dropped");
        return HandlerOutcome::drop();
    }
    tracing::info!(dispute.invoice, dispute.reason, "dispute opened");
    HandlerOutcome::ack()
}

Вручную

src/main.rs

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
struct Dispute {
    invoice: i64,
    reason: String,
}

struct OpenDispute;

impl Handle<Dispute> for OpenDispute {
    fn handle(
        &self,
        dispute: &Dispute,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        if dispute.reason.is_empty() {
            tracing::warn!(dispute.invoice, "dispute without a reason dropped");
            return ready(Err(HandlerOutcome::drop()));
        }
        tracing::info!(dispute.invoice, dispute.reason, "dispute opened");
        ready(Ok(()))
    }
}

SchemaFramed оборачивает JSON-кодек ядра в конверт. Декодирование перешагивает через конверт и читает документ JSON внутри него. Кодирование оформляет значение идентификатором субъекта, который register называет для его типа, так же как с Avro. В сервисе выше роутер даёт подписке disputes этот кодек, потому что побеждает самый точный кодек. JsonSchema выводит JSON Schema типа, которую регистрирует register_json.

Закреплённая локальная схема

Макросы

src/main.rs

fn archive(broker: KafkaBroker) -> impl App {
    let pinned = AvroCodec::local(Invoice::get_schema()).expect("the invoice schema resolves");
    RustStream::new(AppInfo::new("billing-archive", "0.1.0")).with_broker_codec(
        broker,
        pinned,
        |b| {
            b.include(archive_invoice);
        },
    )
}

#[subscriber("invoices.archive")]
async fn archive_invoice(invoice: &Invoice) -> HandlerOutcome {
    tracing::info!(invoice.id, "invoice archived");
    HandlerOutcome::ack()
}

Вручную

src/main.rs

fn archive(broker: KafkaBroker) -> impl App {
    let pinned = AvroCodec::local(Invoice::get_schema()).expect("the invoice schema resolves");
    RustStream::new(AppInfo::new("billing-archive", "0.1.0")).with_broker_codec(
        broker,
        pinned,
        |b| {
            b.include(subscriber("invoices.archive", ArchiveInvoice).build());
        },
    )
}

struct ArchiveInvoice;

impl Handle<Invoice> for ArchiveInvoice {
    fn handle(
        &self,
        invoice: &Invoice,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(invoice.id, "invoice archived");
        ready(Ok(()))
    }
}

AvroCodec::local держит одну схему и применяет её к каждой записи. Тема несёт голые данные Avro: на этом пути нет ни конверта, ни реестра, ни сети. Поэтому сервис архива работает и тестируется без реестра.

Путь с перекодированием

Сервис может и оставить обычные модели serde на JSON-кодеке по умолчанию, а преобразовывать данные на границах брокера. KafkaBroker::schema_registry переводит каждую оформленную доставку в JSON до работы кодека. SchemaFrame, добавленный через RustStream::publish_layer, оформляет каждую исходящую полезную нагрузку JSON в формате, который хранит субъект её адресата: Avro, JSON Schema или Protobuf. Тема без субъекта в реестре публикуется как есть. Этот путь стоит одного преобразования JSON на сообщение и не приводит записи к схеме чтения. Брокер работает по одному из двух путей: при подключённом schema_registry кодек получал бы JSON.

Другой крейт реестра

SchemaRegistry::with_client принимает собственный RegistryClient, например поверх другого HTTP-стека или другого клиентского крейта. Тогда кодеки, предвыборка и путь с перекодированием обращаются к реестру через него.

Крейт, который сам делает всю работу со схемами, например schema_registry_converter, работает с байтами. Обработчик принимает байты записи через тип с #[derive(Deserialized)] и возвращает закодированный ответ через тип с #[derive(Serialized)]. Между таким крейтом и RustStream передаются только байты, поэтому его версия apache-avro не обязана совпадать с версией этого крейта.