Kafka — Basic

Use this as your starting point for Kafka integration. Three OrderCreated messages are published and consumed via a coordinated consumer group, showing how shove's generic Broker\<Kafka\> API maps to rdkafka producers, consumer groups, and partition assignment — including how hold queues are implemented as separate Kafka topics.

Prerequisites

Docker (an Apache Kafka testcontainer is started automatically — no manual docker run needed)
Cargo feature: kafka

Run

cargo run --example kafka_basic --features kafka

Expected output

Published order ORD-0
Published order ORD-1
Published order ORD-2
Processing order ORD-0 ($99.99) [retry=0]
Processing order ORD-1 ($100.99) [retry=0]
Processing order ORD-2 ($101.99) [retry=0]
Done.

Source

//! Basic Kafka publish/consume example.
//!
//! Spins up a Kafka testcontainer automatically (requires a running Docker
//! daemon):
//!
//!     cargo run -q --example kafka_basic --features kafka
 
use std::time::Duration;
 
use serde::{Deserialize, Serialize};
use shove::kafka::{KafkaConfig, KafkaConsumerGroupConfig};
use shove::{
    Broker, ConsumerGroupConfig, Kafka, MessageHandler, MessageMetadata, Outcome, TopologyBuilder,
};
use testcontainers::runners::AsyncRunner;
use testcontainers_modules::kafka::apache::{self, Kafka as KafkaImage};
 
#[derive(Debug, Clone, Serialize, Deserialize)]
struct OrderCreated {
    order_id: String,
    amount: f64,
}
 
shove::define_topic!(
    OrderTopic,
    OrderCreated,
    TopologyBuilder::new("kafka-orders")
        .hold_queue(Duration::from_secs(1))
        .hold_queue(Duration::from_secs(5))
        .dlq()
        .build()
);
 
struct OrderHandler;
 
impl MessageHandler<OrderTopic> for OrderHandler {
    type Context = ();
    async fn handle(&self, message: OrderCreated, metadata: MessageMetadata, _: &()) -> Outcome {
        println!(
            "Processing order {} (${:.2}) [retry={}]",
            message.order_id, message.amount, 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 broker = Broker::<Kafka>::new(KafkaConfig::new(&bootstrap)).await?;
    broker.topology().declare::<OrderTopic>().await?;
 
    // Publish
    let publisher = broker.publisher().await?;
    for i in 0..3 {
        publisher
            .publish::<OrderTopic>(&OrderCreated {
                order_id: format!("ORD-{i}"),
                amount: 99.99 + i as f64,
            })
            .await?;
        println!("Published order ORD-{i}");
    }
 
    // Consume via a coordinated consumer group.
    let mut group = broker.consumer_group();
    group
        .register::<OrderTopic, _>(
            ConsumerGroupConfig::new(KafkaConsumerGroupConfig::new(1..=1)),
            || OrderHandler,
        )
        .await?;
 
    // Stop after 3 s for demo purposes, or on ctrl-c.
    let outcome = group
        .run_until_timeout(
            async {
                tokio::select! {
                    _ = tokio::time::sleep(Duration::from_secs(3)) => {}
                    _ = tokio::signal::ctrl_c() => {}
                }
            },
            Duration::from_secs(10),
        )
        .await;
 
    println!("Done.");
    std::process::exit(outcome.exit_code());
}

Walkthrough

Kafka topic provisioning on declare

broker.topology().declare::<OrderTopic>() creates the Kafka topics that back the topology: one main topic (kafka-orders), two hold-queue topics (kafka-orders-hold-1s, kafka-orders-hold-5s), and a DLQ topic (kafka-orders-dlq). On Kafka, each logical shove queue is a separate Kafka topic. The number of partitions is determined by the consumer group's max_consumers setting — a partition count less than the consumer count would leave some consumers idle.

`KafkaConfig` and bootstrap servers

Broker::<Kafka>::new(KafkaConfig::new(&bootstrap)) initialises the rdkafka producer and consumer configuration pointing at the single-broker testcontainer. In production bootstrap would be a comma-separated list of broker addresses. KafkaConfig exposes additional methods for TLS (with_tls) and SASL authentication (with_sasl) — see the Kafka backend overview for the kafka-ssl feature.

Consumer group with coordinated partition assignment

broker.consumer_group() creates a ConsumerGroup\<Kafka\>. group.register::<OrderTopic, _>(ConsumerGroupConfig::new(KafkaConsumerGroupConfig::new(1..=1)), || OrderHandler) binds the handler factory to OrderTopic with a fixed pool of one consumer. On Kafka, the consumer group registers a standard rdkafka consumer group; when multiple consumers are active they receive distinct partition assignments from the Kafka coordinator — messages within a partition remain ordered.

Hold queues as separate Kafka topics

Unlike RabbitMQ (which uses AMQP dead-letter routing) or NATS (which uses JetStream subjects), Kafka hold queues are separate topics. When a handler returns Outcome::Retry, shove publishes the message to the appropriate hold-queue topic and acknowledges the original. A separate consumer (run internally by the framework) reads from the hold-queue topic after the TTL and republishes to the main topic. This approach keeps messages durable across broker restarts at the cost of one additional produce+consume round-trip per retry.

What to try next

Change new(1..=1) to new(1..=4) and publish 20 messages — the consumer group assigns partitions dynamically; observe rebalance log lines during the 3-second run window.
Return Outcome::Retry from OrderHandler and verify the 1-second hold queue fires and redelivers before the 3-second window closes.
Add TLS by enabling the kafka-ssl feature and configuring KafkaConfig::with_tls — no handler code changes needed.
See Guides: Retries for how hold queues and retry limits interact on Kafka.