Auto

Commit::Auto is librdkafka's timer commit: an ack is advisory, and a retry is not delivered again.

It is the default of every descriptor, and librdkafka's own default. librdkafka stores a record's position as it hands the record to the subscription, and commits the stored position every auto.commit.interval.ms. The handler's outcome changes nothing about that position: an ack and a drop are advisory. A retry reports itself unsupported, and the delivery does not come back on the subscription or on the group's next member. A crash loses the processed records whose position was not committed yet, and skips the stored records the handler had not finished. This suits a consumer that can lose a few records, such as a dashboard or a cache. A consumer that must process every record takes Commit::Tracked.

Example

Macros

examples/kafka_settlement_commit.rs

/// A dashboard can lose a few page views in a crash, so this subscription keeps librdkafka's
/// timer commit. The position is stored as the view is handed over, so a retry asks for
/// something the subscription cannot do: the view is not delivered again.
#[subscriber(KafkaTopic::new("page-views").group("analytics"))]
async fn count_view(view: &PageView, State(dashboard): State<Arc<Dashboard>>) -> HandlerOutcome {
    if let Err(err) = dashboard.count(view) {
        tracing::warn!(%err, "page view lost");
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

Manual

examples/manual/kafka_settlement_commit.rs

/// A dashboard can lose a few page views in a crash, so this subscription keeps librdkafka's
/// timer commit. The position is stored as the view is handed over, so a retry asks for
/// something the subscription cannot do: the view is not delivered again.
struct CountView;

impl Handle<PageView, (), (), (), AppState> for CountView {
    fn handle(
        &self,
        view: &PageView,
        _outs: &(),
        ctx: &mut Context<'_, (), AppState>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        if let Err(err) = ctx.state().dashboard.count(view) {
            tracing::warn!(%err, "page view lost");
            return ready(Err(HandlerOutcome::retry()));
        }
        ready(Ok(()))
    }
}

fn analytics() -> KafkaTopic {
    KafkaTopic::new("page-views").group("analytics")
}