Protobuf в конверте Confluent

Это руководство обрабатывает сообщения Protobuf, сгенерированные prost, в конверте Confluent: читает оформленные записи без реестра и публикует оформленные ответы.

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

Cargo.toml

[dependencies]
ruststream-rdkafka = { version = "0.7", features = ["protobuf"] }
prost = "0.14"

Берите ту версию prost, от которой зависит крейт.

Сгенерированные типы в обработчике

Макросы

src/main.rs

/// What `prost-build` generates for `shipping.Parcel`, plus the decode lane.
#[derive(Clone, PartialEq, prost::Message, Deserialized)]
#[wire(decode = decode_confluent)]
struct Parcel {
    #[prost(int64, tag = "1")]
    id: i64,
    #[prost(string, tag = "2")]
    address: String,
}

/// What `prost-build` generates for `shipping.Label`, plus the encode lane and its topic.
#[derive(Clone, PartialEq, prost::Message, Serialized, Outgoing)]
#[wire(encode = ::prost::Message::encode)]
#[outgoing(name = "labels")]
struct Label {
    #[prost(int64, tag = "1")]
    parcel: i64,
    #[prost(string, tag = "2")]
    address: String,
}

Вручную

src/main.rs

/// What `prost-build` generates for `shipping.Parcel`.
#[derive(Clone, PartialEq, prost::Message)]
struct Parcel {
    #[prost(int64, tag = "1")]
    id: i64,
    #[prost(string, tag = "2")]
    address: String,
}

// The decode lane: the envelope is stepped over, then `prost` reads the message.
impl Deserialized for Parcel {
    type Output<'a> = Self;
    type Error = KafkaError;

    fn from_payload(payload: &[u8]) -> Result<Self, Self::Error> {
        decode_confluent(payload)
    }
}

impl Input for Parcel {
    type Axis = SoloDeserialized<Self>;
}

/// What `prost-build` generates for `shipping.Label`.
#[derive(Clone, PartialEq, prost::Message)]
struct Label {
    #[prost(int64, tag = "1")]
    parcel: i64,
    #[prost(string, tag = "2")]
    address: String,
}

// The encode lane: `prost` writes the message into the publish buffer.
impl Serialized for Label {
    type Error = prost::EncodeError;

    fn wire_bytes(&self, buf: &mut BytesMut) -> Result<WireBytes<'_>, Self::Error> {
        prost::Message::encode(self, &mut *buf)?;
        Ok(WireBytes::InBuffer)
    }
}

impl MessageWire for Label {
    type Wire = SerializedWire;
}

impl ReplyShape for Label {
    type Body = Self;
    type Headers = ();
    type Wire = SerializedReply;
}

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

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

impl MessageHeaders for Label {
    type Contract = NoHeaders;
}

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

Parcel и Label - то, что prost-build генерирует из этой схемы, плюс derive RustStream:

shipping.proto

syntax = "proto3";
package shipping;

message Parcel {
  int64 id = 1;
  string address = 2;
}

message Label {
  int64 parcel = 1;
  string address = 2;
}

В скрипте сборки message_attribute из prost-build добавляет derive к сгенерированным структурам. #[derive(Deserialized)] позволяет обработчику принимать сообщение, а #[derive(Serialized)] - возвращать его. #[wire(..)] называет функцию, которая читает или пишет байты. В ручной форме вместо derive стоят реализации трейтов.

Макросы

src/main.rs

#[subscriber("parcels", reply)]
async fn print_label(parcel: &Parcel) -> Result<Label, HandlerOutcome> {
    if parcel.address.is_empty() {
        tracing::warn!(parcel.id, "parcel without an address dropped");
        return Err(HandlerOutcome::drop());
    }
    tracing::info!(parcel.id, "label printed");
    Ok(Label {
        parcel: parcel.id,
        address: parcel.address.clone(),
    })
}

Вручную

src/main.rs

struct PrintLabel;

impl Handle<Parcel, Label> for PrintLabel {
    fn handle(
        &self,
        parcel: &Parcel,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<Label, HandlerOutcome>> {
        if parcel.address.is_empty() {
            tracing::warn!(parcel.id, "parcel without an address dropped");
            return ready(Err(HandlerOutcome::drop()));
        }
        tracing::info!(parcel.id, "label printed");
        ready(Ok(Label {
            parcel: parcel.id,
            address: parcel.address.clone(),
        }))
    }
}

print_label принимает сгенерированный Parcel и возвращает сгенерированный Label. Посылка без адреса отбрасывается, и для неё ничего не публикуется.

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

decode_confluent - функция декодирования Parcel. Она перешагивает через конверт и путь индексов, а затем сообщение читает prost. Реестр ей не нужен: сгенерированный тип и так знает свои поля. Запись без конверта читается как голое сообщение, поэтому тема, в которую одни продюсеры пишут с конвертом, а другие без, всё равно декодируется.

Публикация оформленных записей

Макросы

src/main.rs

fn service(broker: KafkaBroker, registry: &SchemaRegistry) -> RustStream {
    RustStream::new(AppInfo::new("shipping", "0.1.0")).with_broker(broker, |b| {
        b.include(print_label)
            .out_reply(KafkaPublish::framed(registry));
    })
}

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

Вручную

src/main.rs

fn service(broker: KafkaBroker, registry: &SchemaRegistry) -> RustStream {
    RustStream::new(AppInfo::new("shipping", "0.1.0")).with_broker(broker, |b| {
        b.include(subscriber("parcels", PrintLabel).reply().build())
            .out_reply(KafkaPublish::framed(registry));
    })
}

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

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

Label пишет голый Protobuf, а политика публикации ставит перед ним конверт. Эта политика - KafkaPublish::framed, и для ответа её называет out_reply. При первой публикации в тему она находит субъект темы (labels-value) и кеширует идентификатор его схемы. Путь индексов по умолчанию указывает на первое сообщение схемы. message называет для темы другое сообщение, а subject - другой субъект.

Субъект разрешается при первой публикации, а не при запуске, потому что только тогда известна тема. Если реестр не хранит субъект или хранит под ним другой формат, публикация завершается ошибкой, и доставка завершается повтором. В режиме фиксации по умолчанию повтор не доставляется снова. Режимы описаны в руководстве Фиксация и гарантии доставки.

ProtobufFrame оформляет записи для всего приложения и добавляется через RustStream::publish_layer. Он оформляет то, что уходит через слоты Out и через издателей, которых не называют точки монтирования, если субъект адресата хранит схему Protobuf. Остальное он пропускает как есть. Ответ идёт прямо в собственного издателя и через слой не проходит, поэтому обработчик с ответом называет KafkaPublish::framed. Полезную нагрузку, у которой конверт уже есть, слой не трогает, поэтому слой и политика сочетаются.

Тестирование внутри процесса

Чтению реестр не нужен, поэтому print_label читает оформленные записи под TestApp::start без реестра и без кластера. Первый тест примера публикует оформленную посылку без адреса и проверяет, что она отброшена и что ни одна этикетка не опубликована. Оформленной публикации нужен идентификатор схемы субъекта, поэтому тесты этикетки работают с реестром. Работу с TestApp описывает руководство Тестирование сервиса на Kafka.

Ограничения

Схему Protobuf нельзя вывести из сгенерированного типа. Поэтому исходник .proto регистрируют отдельно: через SchemaRegistry::register с SchemaType::Protobuf или инструментами самого реестра. Схема, зарегистрированная для темы с конвертом, может импортировать только стандартные типы (google/protobuf/*.proto). Другие ссылки на схемы не разрешаются.