Exactly once

KafkaEosPublish commits consumed offsets inside the producer's transaction, so positions move atomically with the records published.

It is the consume-transform-produce shape of Kafka (KIP-447), named EosPublish in the prelude. KafkaEosPublish::new takes the pipeline id, which is also the producer's transactional id. Each source subscription names the same id in Commit::Transactional: its consumer stops committing on its own, and the pipeline commits its offsets. One pipeline serves every lane of a workers(n, by_key) pool: publishes join the open transaction concurrently, and only the commit is exclusive.

The pipeline works in windows. Every commit_interval (100 ms by default) it waits until the window's deliveries have settled, adds their positions and the group's metadata to the transaction, and commits. A record becomes visible at that commit, so end-to-end latency is at least the interval. A failed publish, a failed commit or a participant that does not settle in time aborts the window. The consumers then seek back to the last committed offsets, and the whole window is delivered and published again in a fresh transaction, so the output topic never holds a duplicate. A retry() from a participant holds its window until the transaction deadline, transaction_timeout. The retry_after copy does not apply to the pipeline's replies: a delayed copy would separate the record from its offset. The window's commit runs on the runtime the broker connected on, so a handler on a dedicated thread may open a window.

The transactional commit needs consumer group membership. So Commit::Transactional is a KafkaTopic or KafkaTopics mode; KafkaPartitions, which joins no group, works with Commit::Auto and Commit::Tracked. A replying handler reaches the window through EosReplies, and a handler that publishes by hand through EosPipeline.

Example

An order is priced, and the priced order commits together with the offset of the order it answers.

Macros
Manual

examples/kafka_eos.rs
/// Prices an order; the subscription commits its offsets through the `pricing-1` pipeline.
#[subscriber(
    KafkaTopic::new("raw-orders").commit(Commit::Transactional("pricing-1".into())),
    reply
)]
async fn price(order: &Order) -> PricedOrder {
    PricedOrder {
        id: order.id,
        cents: order.cents,
        tax_cents: order.cents / 5,
    }
}

examples/manual/kafka_eos.rs
/// Prices an order; the subscription commits its offsets through the `pricing-1` pipeline.
struct Price;

impl Handle<Order, PricedOrder> for Price {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<PricedOrder, HandlerOutcome>> {
        ready(Ok(PricedOrder {
            id: order.id,
            cents: order.cents,
            tax_cents: order.cents / 5,
        }))
    }
}

Macros
Manual

examples/kafka_eos.rs
fn app() -> impl App {
    // The policy's id is the pipeline the subscription names, and the producer's transactional id.
    let pipeline = EosPublish::new("pricing-1").commit_interval(Duration::from_millis(100));
    RustStream::new(AppInfo::new("pricing", "0.1.0")).with_broker(
        KafkaBroker::new(["localhost:9092"]).default_group("pricing-svc"),
        |b| {
            b.include(price).out_reply(pipeline).transform(EosReplies);
        },
    )
}

examples/manual/kafka_eos.rs
fn app() -> impl App {
    // The policy's id is the pipeline the subscription names, and the producer's transactional id.
    let pipeline = EosPublish::new("pricing-1").commit_interval(Duration::from_millis(100));
    RustStream::new(AppInfo::new("pricing", "0.1.0")).with_broker(
        KafkaBroker::new(["localhost:9092"]).default_group("pricing-svc"),
        |b| {
            let orders = subscriber("raw-orders", Price)
                .map_source(|_| {
                    KafkaTopic::new("raw-orders").commit(Commit::Transactional("pricing-1".into()))
                })
                .reply();
            b.include(orders.build())
                .out_reply(pipeline)
                .transform(EosReplies);
        },
    )
}

Example​

Example