Avro transcode

On a broker with schema_registry an Avro datum arrives as JSON, and SchemaFrame writes Avro where the reply's subject holds an Avro schema.

The handler and its models stay on the default JSON codec and carry no Avro type. On the way in, the subscription reads the envelope, resolves the writer schema by its id and rewrites the payload as a JSON document. On the way out, the layer resolves the subject of the destination topic, {topic}-value, and writes the JSON payload as an Avro datum of that subject's schema, inside the envelope. The reply's model derives AvroSchema only to register its subject at startup with register_avro. The flow is the same as the JSON transcode; the subject's schema type decides the format on the wire.

The transcode is the compatibility path for a service that keeps plain serde models on a registry-backed topic. Each message takes a hop through JSON. A datum is read with its writer schema alone, so a field the writer never wrote has no default to take; the codec with resolve_onto is the path that resolves schemas. An Avro type that JSON has no form for does not pass through the transcode.

Example

Macros
Manual

examples/kafka_avro.rs
#[derive(Debug, PartialEq, Serialize, Deserialize)]
struct Ticket {
    id: i64,
    seat: String,
}

// `AvroSchema` gives the type the schema its subject is registered with.
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing, AvroSchema)]
#[outgoing(name = "ticket-confirmations")]
struct Confirmation {
    id: i64,
    accepted: bool,
}

// An ordinary handler on the default JSON codec: the broker turned the Avro datum into JSON.
#[subscriber("tickets", reply)]
async fn confirm(ticket: &Ticket) -> Confirmation {
    tracing::info!(ticket.id, ticket.seat, "ticket confirmed");
    Confirmation {
        id: ticket.id,
        accepted: true,
    }
}

fn app(registry: SchemaRegistry) -> impl App {
    let broker = KafkaBroker::new(["localhost:9092"])
        .default_group("tickets-svc")
        .schema_registry(registry.clone());

    RustStream::new(AppInfo::new("tickets", "0.1.0"))
        // The reply's subject holds an Avro schema, so the layer writes the reply as Avro.
        .publish_layer(SchemaFrame::new(registry.clone()))
        .on_startup(async move |()| {
            registry
                .register_avro::<Confirmation>("ticket-confirmations-value")
                .await?;
            Ok::<_, KafkaError>(())
        })
        .with_broker(broker, |b| {
            b.include(confirm);
        })
}

#[ruststream::app]
fn service() -> impl App {
    app(SchemaRegistry::new("http://localhost:8081"))
}

examples/manual/kafka_avro.rs
#[derive(Debug, PartialEq, Serialize, Deserialize, JsonSchema)]
struct Ticket {
    id: i64,
    seat: String,
}

// `AvroSchema` gives the type the schema its subject is registered with.
#[derive(Debug, PartialEq, Serialize, Deserialize, JsonSchema, AvroSchema)]
struct Confirmation {
    id: i64,
    accepted: bool,
}

// `#[derive(Outgoing)]` with `#[outgoing(name = "ticket-confirmations")]`.
impl OutgoingDestination for Confirmation {
    type Form = FixedName;
    const DESTINATION: &'static str = "ticket-confirmations";
}

impl MessageHeaders for Confirmation {
    type Contract = NoHeaders;
}

/// The definition `#[subscriber("tickets", reply)]` would have minted.
struct Confirm;

impl Handle<Ticket, Confirmation> for Confirm {
    // An ordinary handler on the default JSON codec: the broker turned the Avro datum into JSON.
    fn handle(
        &self,
        ticket: &Ticket,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<Confirmation, HandlerOutcome>> {
        tracing::info!(ticket.id, ticket.seat, "ticket confirmed");
        ready(Ok(Confirmation {
            id: ticket.id,
            accepted: true,
        }))
    }
}

fn app(registry: SchemaRegistry) -> impl App {
    let broker = KafkaBroker::new(["localhost:9092"])
        .default_group("tickets-svc")
        .schema_registry(registry.clone());

    RustStream::new(AppInfo::new("tickets", "0.1.0"))
        // The reply's subject holds an Avro schema, so the layer writes the reply as Avro.
        .publish_layer(SchemaFrame::new(registry.clone()))
        .on_startup(async move |()| {
            registry
                .register_avro::<Confirmation>("ticket-confirmations-value")
                .await?;
            Ok::<_, KafkaError>(())
        })
        .with_broker(broker, |b| {
            b.include(subscriber("tickets", Confirm).reply().build());
        })
}

// What `#[ruststream::app]` writes: the command line on a multi-threaded Tokio runtime.
fn main() -> ExitCode {
    run_main(|| app(SchemaRegistry::new("http://localhost:8081")))
}

Example​

Example