Kafka — Sequenced

Use this when you need per-entity FIFO ordering on Kafka. Events for two users are published in interleaved order; each user's events arrive strictly in sequence. The example also highlights the key Kafka constraint: partition count caps the maximum number of active consumers, so partition count must be planned before topic creation.

Prerequisites

Docker (an Apache Kafka testcontainer is started automatically)
Cargo feature: kafka

Run

cargo run --example kafka_sequenced --features kafka

Expected output

Events for each user arrive in seq order, but events for different users may interleave across shards:

Published 10 events (5 per user)
[user=alice] action=action-0 seq=0 (retry=0)
[user=bob] action=action-0 seq=0 (retry=0)
[user=alice] action=action-1 seq=1 (retry=0)
[user=bob] action=action-1 seq=1 (retry=0)
...
Done.

Source

//! Sequenced Kafka example — per-key ordering.
//!
//! Spins up a Kafka testcontainer automatically (requires a running Docker
//! daemon):
//!
//!     cargo run -q --example kafka_sequenced --features kafka
//!
//! Note: per-key FIFO consumption (`run_fifo`) isn't yet surfaced on the
//! generic `Broker<B>` / `ConsumerSupervisor<B>` / `ConsumerGroup<B>`
//! wrappers — this example therefore keeps using the backend-specific
//! `KafkaConsumer::run_fifo` directly.
 
use std::time::Duration;
 
use serde::{Deserialize, Serialize};
use shove::kafka::{
    KafkaClient, KafkaConfig, KafkaConsumer, KafkaPublisher, KafkaTopologyDeclarer,
};
use shove::{
    ConsumerOptions, Kafka, MessageHandler, MessageMetadata, Outcome, SequenceFailure,
    SequencedTopic, Topic, TopologyBuilder,
};
use testcontainers::runners::AsyncRunner;
use testcontainers_modules::kafka::apache::{self, Kafka as KafkaImage};
use tokio_util::sync::CancellationToken;
 
#[derive(Debug, Clone, Serialize, Deserialize)]
struct UserEvent {
    user_id: String,
    action: String,
    seq: u32,
}
 
shove::define_sequenced_topic!(
    UserEventTopic,
    UserEvent,
    |msg: &UserEvent| msg.user_id.clone(),
    TopologyBuilder::new("kafka-user-events")
        .sequenced(SequenceFailure::Skip)
        .routing_shards(4)
        .hold_queue(Duration::from_secs(1))
        .dlq()
        .build()
);
 
struct UserEventHandler;
 
impl MessageHandler<UserEventTopic> for UserEventHandler {
    type Context = ();
    async fn handle(&self, message: UserEvent, metadata: MessageMetadata, _: &()) -> Outcome {
        println!(
            "[user={}] action={} seq={} (retry={})",
            message.user_id, message.action, message.seq, metadata.retry_count
        );
        Outcome::Ack
    }
}
 
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    tracing_subscriber::fmt::init();
 
    let container = KafkaImage::default().start().await?;
    let port = container.get_host_port_ipv4(apache::KAFKA_PORT).await?;
    let bootstrap = format!("127.0.0.1:{port}");
 
    let config = KafkaConfig::new(&bootstrap);
    let client = KafkaClient::connect(&config).await?;
 
    // Declare topology
    let declarer = KafkaTopologyDeclarer::new(client.clone());
    declarer.declare(UserEventTopic::topology()).await?;
 
    // Publish events for two users
    let publisher = KafkaPublisher::new(client.clone()).await?;
    for seq in 0..5u32 {
        for user in &["alice", "bob"] {
            publisher
                .publish::<UserEventTopic>(&UserEvent {
                    user_id: user.to_string(),
                    action: format!("action-{seq}"),
                    seq,
                })
                .await?;
        }
    }
    println!("Published 10 events (5 per user)");
 
    // Consume with FIFO ordering
    let shutdown = CancellationToken::new();
    let shutdown_clone = shutdown.clone();
    tokio::spawn(async move {
        tokio::time::sleep(Duration::from_secs(5)).await;
        shutdown_clone.cancel();
    });
 
    let consumer = KafkaConsumer::new(client.clone());
    let options = ConsumerOptions::<Kafka>::new().with_shutdown(shutdown);
    consumer
        .run_fifo::<UserEventTopic, _>(UserEventHandler, (), options)
        .await?;
 
    client.shutdown().await;
    println!("Done.");
    drop(container);
    Ok(())
}

Walkthrough

Partition-key routing on Kafka

UserEventTopic uses .routing_shards(4) which on Kafka means 4 Kafka partitions on the main topic. The key function |msg: &UserEvent| msg.user_id.clone() is hashed (consistent hashing) to select a partition; all events for the same user always land on the same partition. Within a Kafka partition messages are stored in strict append order — this is the mechanism that gives shove its per-key ordering guarantee. Changing the partition count requires deleting and recreating the topic; Kafka can only expand, not shrink.

`KafkaConsumer::run_fifo` and the partition constraint

KafkaConsumer::new(client).run_fifo::<UserEventTopic, _>(handler, ctx, opts) is the backend-specific path for per-key FIFO (not yet exposed on the generic ConsumerGroup\<Kafka\> API). With 4 partitions and 1 consumer, the consumer owns all 4 partitions. Adding a second consumer triggers a Kafka rebalance and each consumer is assigned 2 partitions. If the consumer count exceeds the partition count, some consumers receive no partitions and sit idle — a Kafka-specific constraint that does not apply to AMQP or JetStream backends.

`SequenceFailure::Skip` and DLQ routing

.sequenced(SequenceFailure::Skip) means that if a message for alice is rejected, only that message goes to the DLQ; subsequent messages for alice continue to be processed. The rejected message is published to the DLQ topic by the framework before the consumer commits its offset, ensuring at-least-once delivery for the DLQ entry even across consumer restarts.

Cancellation token shutdown

CancellationToken::new() is cancelled after 5 seconds by a spawned task. ConsumerOptions::<Kafka>::new().with_shutdown(shutdown) passes the token to run_fifo, which stops polling and drains when the token fires. client.shutdown().await then closes the rdkafka connections cleanly.

What to try next

Change routing_shards(4) to routing_shards(8) and consumers to 8 in KafkaConsumerGroupConfig — each partition gets exactly one consumer, maximising parallelism with no idle consumers.
Switch SequenceFailure::Skip to SequenceFailure::FailAll and make the handler reject alice seq=2 — all subsequent messages for Alice are DLQ'd while Bob's sequence continues.
Increase the consumer count beyond the partition count — observe that the extra consumers sit idle (no partitions assigned) and log a warning.
Read Guides: Sequenced for a full explanation of shard routing and failure modes across backends.