Passthrough

A descriptor's config passes any consumer property; one the commit mode owns is refused at startup, naming the property.

KafkaTopic::config, and the same method on KafkaTopics and KafkaPartitions, sets one librdkafka consumer property by its name. It reaches the properties the descriptor has no typed option for: fetch sizes, queue limits, timeouts. The passthrough applies last, so it wins over the typed options and over the broker-wide configuration. A property the commit mode owns is the exception. Under Commit::Tracked and Commit::Transactional the crate computes the committed position itself, through enable.auto.offset.store and enable.auto.commit. A passthrough that sets either one would leave the position to librdkafka while every ack decided nothing. So the subscription refuses to open, and the startup error names the subscription and the property. Commit::Auto owns neither property and takes both from the passthrough. A property for every consumer of the broker goes on KafkaBroker instead.

Example

fetch.queue.backoff.ms shortens the pause librdkafka takes when a partition's local queue is full, from a second to 10 ms.

Macros
Manual

examples/kafka_subscriptions_passthrough.rs
/// Settles payments; librdkafka pauses a full partition for 10 ms instead of a second.
#[subscriber(
    KafkaTopic::new("payments")
        .group("payments-svc")
        .commit(Commit::Tracked)
        .config("fetch.queue.backoff.ms", "10")
)]
async fn settle(payment: &Payment) -> HandlerOutcome {
    tracing::info!(payment.id, "settled");
    HandlerOutcome::ack()
}

examples/manual/kafka_subscriptions_passthrough.rs
/// Settles every payment it reads.
struct Settle;

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

fn app() -> impl App<State = ()> {
    let broker = KafkaBroker::new(["localhost:9092"]);
    RustStream::new(AppInfo::new("payments", "0.1.0")).with_broker(broker, |b| {
        // librdkafka pauses a full partition for 10 ms instead of a second.
        let payments = KafkaTopic::new("payments")
            .group("payments-svc")
            .commit(Commit::Tracked)
            .config("fetch.queue.backoff.ms", "10");
        b.include(
            subscriber("payments", Settle)
                .build()
                .map_source(|_| payments),
        );
    })
}

Example​

Example