Durability

Durability is the producer's acks; an idempotent producer with Commit::Tracked gives at-least-once from input to output.

producer_config sets a librdkafka producer property on the broker, for every publisher of the service. acks says which replicas confirm a record before a publish returns: all waits for every in-sync replica. enable.idempotence makes the producer number its sends, so the cluster writes a record once even when the producer sends it again. librdkafka checks the values when the broker connects, so a wrong value stops the service at startup. A handler that replies settles its input after the reply's publish returned. Under Commit::Tracked the input's offset is committed only then, so a crash before it brings the input back and the reply is sent again. A reader of the output can therefore see a reply twice; exactly once takes a transactional pipeline.

Example

Macros
Manual

examples/kafka_publishing_durability.rs
/// The input's offset is committed once the receipt is acknowledged by every in-sync replica,
/// so a crash in between brings the payment back instead of losing the receipt.
#[subscriber(KafkaTopic::new("payments").group("receipts-svc").commit(Commit::Tracked), reply)]
async fn issue_receipt(payment: &Payment) -> Receipt {
    Receipt {
        payment_id: payment.id,
    }
}

#[ruststream::app]
fn app() -> impl App {
    // Every in-sync replica confirms a record, and a retried send is written once.
    let broker = KafkaBroker::new(["localhost:9092"])
        .producer_config("acks", "all")
        .producer_config("enable.idempotence", "true");
    RustStream::new(AppInfo::new("receipts", "0.1.0")).with_broker(broker, |b| {
        b.include(issue_receipt).out_reply(Publish::default());
    })
}

examples/manual/kafka_publishing_durability.rs
/// The input's offset is committed once the receipt is acknowledged by every in-sync replica,
/// so a crash in between brings the payment back instead of losing the receipt.
struct IssueReceipt;

impl Handle<Payment, Receipt> for IssueReceipt {
    fn handle(
        &self,
        payment: &Payment,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<Receipt, HandlerOutcome>> {
        ready(Ok(Receipt {
            payment_id: payment.id,
        }))
    }
}

fn payments() -> KafkaTopic {
    KafkaTopic::new("payments")
        .group("receipts-svc")
        .commit(Commit::Tracked)
}

fn app() -> impl App {
    // Every in-sync replica confirms a record, and a retried send is written once.
    let broker = KafkaBroker::new(["localhost:9092"])
        .producer_config("acks", "all")
        .producer_config("enable.idempotence", "true");
    RustStream::new(AppInfo::new("receipts", "0.1.0")).with_broker(broker, |b| {
        let issue = subscriber("payments", IssueReceipt).map_source(|_| payments());
        b.include(issue.reply().build())
            .out_reply(Publish::default());
    })
}

Example​

Example