Tracked

Commit::Tracked commits below the lowest unsettled delivery, so a retry comes back and nothing is committed past it.

The subscription stores a position only when a handler settles a delivery. The stored position is the offset just below the lowest delivery still in flight, so handlers that finish out of order, on several workers, never commit past an unfinished one. librdkafka's timer commits the stored position in the background and once more when the consumer closes. An ack and a drop settle the offset. A retry hands the delivery back to its subscription, which delivers it again before the records behind it. Until it is settled, the group's committed position stays below it, so a crash or a rebalance brings it to the next member. Offsets the consumer never receives, such as transaction markers and compacted records, do not hold the position back. Tracked with an idempotent producer gives at-least-once from input to output, as Durability shows.

Example

Macros
Manual

examples/kafka_settlement_commit.rs
/// A charge the gateway turned away comes back on this subscription before anything behind it,
/// and the group commits nothing past it until it is settled.
#[subscriber(KafkaTopic::new("charges").group("payments").commit(Commit::Tracked))]
async fn take_charge(charge: &Charge, State(gateway): State<Arc<Gateway>>) -> HandlerOutcome {
    if let Err(err) = gateway.charge(charge) {
        tracing::warn!(%err, "charge retried");
        return HandlerOutcome::retry();
    }
    tracing::info!(charge.id, "charge taken");
    HandlerOutcome::ack()
}

examples/manual/kafka_settlement_commit.rs
/// A charge the gateway turned away comes back on this subscription before anything behind it,
/// and the group commits nothing past it until it is settled.
struct TakeCharge;

impl Handle<Charge, (), (), (), AppState> for TakeCharge {
    fn handle(
        &self,
        charge: &Charge,
        _outs: &(),
        ctx: &mut Context<'_, (), AppState>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        if let Err(err) = ctx.state().gateway.charge(charge) {
            tracing::warn!(%err, "charge retried");
            return ready(Err(HandlerOutcome::retry()));
        }
        tracing::info!(charge.id, "charge taken");
        ready(Ok(()))
    }
}

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

Example​

Example