MissingSubject

MissingSubject decides what a codec does when the registry no longer holds its subject: refuse, register the type's schema, or publish unframed.

The policy is set on the prefetch with on_missing_subject before codecs are built from it, and it runs when the broker connects. Refuse, the default, stops the app with an error that names the subject: a registry is a shared contract, and creating a subject in it belongs to whoever owns the topic. AutoRegister registers the schema the type carries under the subject and logs a warning; it suits a producer that owns its subjects. PublishUnframed writes the bare datum and logs a warning, for a topic that must outlive its registry; a registry-backed consumer cannot read those records.

The two deletions of a subject differ. A soft delete hides the subject while its ids still resolve, so consumers keep reading and only the producer meets the policy. A permanent delete removes the ids too, and a record that names one of them no longer decodes anywhere. Registering the schema again afterwards mints a new id.

The transcode reads a missing subject as a plain topic and publishes JSON there. Naming a codec declares its topic registry-backed, so the policy belongs to the codecs.

Example

Macros
Manual

examples/kafka_registry_missing.rs
#[derive(Debug, PartialEq, Serialize, Deserialize, AvroSchema)]
struct Order {
    id: i64,
}

#[derive(Debug, PartialEq, Serialize, Deserialize, AvroSchema, Outgoing)]
#[outgoing(name = "parcels")]
struct Parcel {
    order: i64,
    weight_grams: i64,
}

#[subscriber("orders", reply)]
async fn pack(order: &Order) -> Parcel {
    tracing::info!(order.id, "order packed");
    Parcel {
        order: order.id,
        weight_grams: 500,
    }
}

fn app(registry: SchemaRegistry) -> impl App {
    // This service owns the parcels subject, so a missing one is put back at connect.
    let prefetch = SchemaPrefetch::new(registry).on_missing_subject(MissingSubject::AutoRegister);
    let codec = AvroCodec::registry(&prefetch).register::<Parcel>("parcels-value");
    let broker = KafkaBroker::new(["localhost:9092"])
        .default_group("packing-svc")
        .schema_prefetch(prefetch);

    let replies = codec.clone();
    RustStream::new(AppInfo::new("packing", "0.1.0")).with_broker_codec(broker, codec, |b| {
        b.include(pack).out_reply(Publish::default()).codec(replies);
    })
}

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

examples/manual/kafka_registry_missing.rs
#[derive(Debug, PartialEq, Serialize, Deserialize, AvroSchema, JsonSchema)]
struct Order {
    id: i64,
}

#[derive(Debug, PartialEq, Serialize, Deserialize, AvroSchema, JsonSchema)]
struct Parcel {
    order: i64,
    weight_grams: i64,
}

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

impl MessageHeaders for Parcel {
    type Contract = NoHeaders;
}

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

impl Handle<Order, Parcel> for Pack {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<Parcel, HandlerOutcome>> {
        tracing::info!(order.id, "order packed");
        ready(Ok(Parcel {
            order: order.id,
            weight_grams: 500,
        }))
    }
}

fn app(registry: SchemaRegistry) -> impl App {
    // This service owns the parcels subject, so a missing one is put back at connect.
    let prefetch = SchemaPrefetch::new(registry).on_missing_subject(MissingSubject::AutoRegister);
    let codec = AvroCodec::registry(&prefetch).register::<Parcel>("parcels-value");
    let broker = KafkaBroker::new(["localhost:9092"])
        .default_group("packing-svc")
        .schema_prefetch(prefetch);

    let replies = codec.clone();
    RustStream::new(AppInfo::new("packing", "0.1.0")).with_broker_codec(broker, codec, |b| {
        b.include(subscriber("orders", Pack).reply().build())
            .out_reply(Publish::default())
            .codec(replies);
    })
}

// 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