Messaging and Kafka

A message broker moves events between services without the sender waiting for the receiver. This lesson uses Kafka with the rdkafka crate to publish Storefront order events and to consume them with an idempotent handler and a dead-letter topic.

Core terms

Queue: a broker deletes a message after a consumer acknowledges it. Each message goes to one consumer.
Log: a broker keeps messages for a retention period. Each consumer group tracks its own position (offset), so several groups read the same messages and can replay.
Topic: a named log. It is split into partitions, each an ordered, append-only sequence.
Offset: the position of a message inside one partition.
Key: bytes chosen by the producer. Equal keys always go to the same partition.
Consumer group: consumers that share a group id divide the partitions. One partition is read by one consumer of a group at a time. A second group receives every message again.
Ordering: guaranteed inside one partition only. Using the order id as key keeps all events of one order in sequence.
Rebalance: partitions move between consumers when a consumer joins or leaves. Moved partitions pause briefly.
Lag: the distance between the latest offset and the group's committed offset. It is the main health signal of a consumer.

Delivery semantics depend on when the offset is committed:

At-most-once: commit before handling. A crash loses the message.
At-least-once: commit after handling. A crash repeats the message, so handlers must tolerate duplicates.
Exactly-once effect: at-least-once delivery plus an idempotent handler, such as a processed-events table updated in the same transaction as the effect.

Project setup

Start Kafka (single node, KRaft mode) and Postgres:

Terminal

docker run -d --name kafka -p 9092:9092 apache/kafka:3.9.0
docker run -d --name pg -e POSTGRES_PASSWORD=postgres -e POSTGRES_DB=storefront -p 5432:5432 postgres:16

Create the topics with three partitions, then the tables:

Terminal

docker exec kafka /opt/kafka/bin/kafka-topics.sh --create --topic orders \
  --partitions 3 --bootstrap-server localhost:9092
docker exec kafka /opt/kafka/bin/kafka-topics.sh --create --topic orders.dlq \
  --partitions 1 --bootstrap-server localhost:9092
docker exec -i pg psql -U postgres -d storefront < migrations/0001_init.sql

migrations/0001_init.sql

SQL

CREATE TABLE processed_events (
    event_id TEXT PRIMARY KEY,
    processed_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE TABLE shipments (
    id BIGSERIAL PRIMARY KEY,
    order_id BIGINT NOT NULL,
    status TEXT NOT NULL
);

Cargo.toml

TOML

[package]
name = "storefront-events"
version = "0.1.0"
edition = "2021"

[dependencies]
rdkafka = { version = "0.36", features = ["cmake-build"] }
tokio = { version = "1", features = ["full"] }
serde = { version = "1", features = ["derive"] }
serde_json = "1"
sqlx = { version = "0.8", features = ["runtime-tokio", "postgres"] }

The cmake-build feature compiles librdkafka from source, so cmake and a C compiler must be installed.

Example 1: The event type and a keyed producer

Producer settings acks=all (wait for all in-sync replicas) and enable.idempotence=true (the broker drops retry duplicates) prevent loss and duplicates on the producer side.

src/lib.rs

Rust

use serde::{Deserialize, Serialize};

#[derive(Debug, Serialize, Deserialize)]
pub struct OrderEvent {
    pub event_id: String,
    pub order_id: i64,
    pub status: String,
}

src/bin/producer.rs

Rust

use rdkafka::config::ClientConfig;
use rdkafka::producer::{FutureProducer, FutureRecord};
use std::time::Duration;
use storefront_events::OrderEvent;

#[tokio::main]
async fn main() {
    let producer: FutureProducer = ClientConfig::new()
        .set("bootstrap.servers", "localhost:9092")
        .set("acks", "all")
        .set("enable.idempotence", "true")
        .create()
        .expect("producer");

    let statuses = ["placed", "paid", "shipped"];
    for (index, status) in statuses.iter().enumerate() {
        let event = OrderEvent {
            event_id: format!("7-{index}"),
            order_id: 7,
            status: status.to_string(),
        };
        let payload = serde_json::to_string(&event).unwrap();
        let key = event.order_id.to_string();
        producer
            .send(
                FutureRecord::to("orders").key(&key).payload(&payload),
                Duration::from_secs(5),
            )
            .await
            .expect("delivery failed");
    }
    println!("sent {} events", statuses.len());
}

send returns a future that resolves when the broker acknowledges the record. The Duration bounds how long the record waits in the local queue.
All three events use key 7, so they land in one partition and keep their order.

Output

sent 3 events

Example 2: A consumer with manual commits

With enable.auto.commit=false, the program commits an offset only after handling the message, which gives at-least-once delivery. auto.offset.reset=earliest starts a new group at the beginning of the topic.

src/bin/consumer.rs

Rust

use rdkafka::config::ClientConfig;
use rdkafka::consumer::{CommitMode, Consumer, StreamConsumer};
use rdkafka::Message;
use storefront_events::OrderEvent;

#[tokio::main]
async fn main() {
    let consumer: StreamConsumer = ClientConfig::new()
        .set("bootstrap.servers", "localhost:9092")
        .set("group.id", "shipping")
        .set("enable.auto.commit", "false")
        .set("auto.offset.reset", "earliest")
        .create()
        .expect("consumer");
    consumer.subscribe(&["orders"]).expect("subscribe");

    loop {
        let message = consumer.recv().await.expect("recv");
        if let Some(payload) = message.payload() {
            let event: OrderEvent = serde_json::from_slice(payload).expect("json");
            println!("order {} is {}", event.order_id, event.status);
        }
        consumer
            .commit_message(&message, CommitMode::Async)
            .expect("commit");
    }
}

After Example 1 ran once, the consumer prints the three events in order (stop it with Ctrl-C):

Output

order 7 is placed
order 7 is paid
order 7 is shipped

With events for several orders, lines of one order stay in order and lines of different orders may interleave, because different keys may sit in different partitions. Running a second copy of this program with the same group id splits the three partitions between the two copies. Parallelism is capped by the partition count, so a fourth consumer would stay idle.

Example 3: Idempotent handling, retries, and a dead-letter topic

A message that cannot be parsed will never succeed, and one that fails on a database error may succeed later. The worker sends unparsable messages straight to orders.dlq and retries database failures three times before doing the same. Committing after the dead-letter send keeps one bad message from blocking its partition.

The handler inserts the event id into processed_events with ON CONFLICT DO NOTHING. If no row was inserted, the event is a duplicate and the effect is skipped. The id insert and the effect share one transaction, so both happen or neither does. Transactions are covered in SQL transactions.

src/bin/worker.rs

Rust

use rdkafka::config::ClientConfig;
use rdkafka::consumer::{CommitMode, Consumer, StreamConsumer};
use rdkafka::producer::{FutureProducer, FutureRecord};
use rdkafka::Message;
use sqlx::postgres::PgPoolOptions;
use sqlx::PgPool;
use std::time::Duration;
use storefront_events::OrderEvent;

async fn apply(pool: &PgPool, event: &OrderEvent) -> Result<bool, sqlx::Error> {
    let mut tx = pool.begin().await?;
    let inserted = sqlx::query(
        "INSERT INTO processed_events (event_id) VALUES ($1) ON CONFLICT DO NOTHING",
    )
    .bind(&event.event_id)
    .execute(&mut *tx)
    .await?
    .rows_affected();

    if inserted == 1 {
        sqlx::query("INSERT INTO shipments (order_id, status) VALUES ($1, $2)")
            .bind(event.order_id)
            .bind(&event.status)
            .execute(&mut *tx)
            .await?;
    }
    tx.commit().await?;
    Ok(inserted == 1)
}

#[tokio::main]
async fn main() {
    let pool = PgPoolOptions::new()
        .max_connections(5)
        .connect("postgres://postgres:postgres@localhost:5432/storefront")
        .await
        .expect("database");

    let consumer: StreamConsumer = ClientConfig::new()
        .set("bootstrap.servers", "localhost:9092")
        .set("group.id", "shipments-worker")
        .set("enable.auto.commit", "false")
        .set("auto.offset.reset", "earliest")
        .create()
        .expect("consumer");
    consumer.subscribe(&["orders"]).expect("subscribe");

    let producer: FutureProducer = ClientConfig::new()
        .set("bootstrap.servers", "localhost:9092")
        .set("acks", "all")
        .create()
        .expect("producer");

    loop {
        let message = consumer.recv().await.expect("recv");
        let payload = message.payload().unwrap_or_default();

        let mut handled = false;
        if let Ok(event) = serde_json::from_slice::<OrderEvent>(payload) {
            for attempt in 1..=3u64 {
                match apply(&pool, &event).await {
                    Ok(true) => println!("applied {}", event.event_id),
                    Ok(false) => println!("duplicate {}", event.event_id),
                    Err(error) => {
                        eprintln!("attempt {attempt} failed: {error}");
                        tokio::time::sleep(Duration::from_millis(200 * attempt)).await;
                        continue;
                    }
                }
                handled = true;
                break;
            }
        }

        if !handled {
            let record = FutureRecord::to("orders.dlq")
                .key(message.key().unwrap_or_default())
                .payload(payload);
            producer
                .send(record, Duration::from_secs(5))
                .await
                .expect("dead-letter send failed");
            println!("dead-lettered");
        }

        consumer
            .commit_message(&message, CommitMode::Sync)
            .expect("commit");
    }
}

Retries wait 200, 400, then 600 milliseconds. Fixed attempts keep the worker from looping forever.
The dead-letter record keeps the original key and payload so it can be inspected or replayed.
Watch lag with docker exec kafka /opt/kafka/bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group shipments-worker.

Partitions, retention, and replication

Partition count caps consumer parallelism and can only grow. Growing it changes which partition a key maps to, which breaks per-key order across the change.
Hot key: one key with far more traffic than others overloads one partition.
Retention (retention.ms, retention.bytes) decides how far back a new group or a replay can read.
Replication factor and min.insync.replicas decide how many broker failures the topic survives. acks=all waits for the in-sync set.

Complete example

Text

storefront-events/
  Cargo.toml
  migrations/0001_init.sql
  src/
    lib.rs
    bin/producer.rs
    bin/consumer.rs
    bin/worker.rs

Files Cargo.toml, migrations/0001_init.sql, src/lib.rs, src/bin/producer.rs, src/bin/consumer.rs, and src/bin/worker.rs are exactly the ones shown above in Project setup and Examples 1 to 3. The infrastructure setup is repeated for a run from scratch.

Cargo.toml

TOML

[package]
name = "storefront-events"
version = "0.1.0"
edition = "2021"

[dependencies]
rdkafka = { version = "0.36", features = ["cmake-build"] }
tokio = { version = "1", features = ["full"] }
serde = { version = "1", features = ["derive"] }
serde_json = "1"
sqlx = { version = "0.8", features = ["runtime-tokio", "postgres"] }

migrations/0001_init.sql

SQL

CREATE TABLE processed_events (
    event_id TEXT PRIMARY KEY,
    processed_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE TABLE shipments (
    id BIGSERIAL PRIMARY KEY,
    order_id BIGINT NOT NULL,
    status TEXT NOT NULL
);

src/lib.rs

Rust

use serde::{Deserialize, Serialize};

#[derive(Debug, Serialize, Deserialize)]
pub struct OrderEvent {
    pub event_id: String,
    pub order_id: i64,
    pub status: String,
}

src/bin/producer.rs

Rust

use rdkafka::config::ClientConfig;
use rdkafka::producer::{FutureProducer, FutureRecord};
use std::time::Duration;
use storefront_events::OrderEvent;

#[tokio::main]
async fn main() {
    let producer: FutureProducer = ClientConfig::new()
        .set("bootstrap.servers", "localhost:9092")
        .set("acks", "all")
        .set("enable.idempotence", "true")
        .create()
        .expect("producer");

    let statuses = ["placed", "paid", "shipped"];
    for (index, status) in statuses.iter().enumerate() {
        let event = OrderEvent {
            event_id: format!("7-{index}"),
            order_id: 7,
            status: status.to_string(),
        };
        let payload = serde_json::to_string(&event).unwrap();
        let key = event.order_id.to_string();
        producer
            .send(
                FutureRecord::to("orders").key(&key).payload(&payload),
                Duration::from_secs(5),
            )
            .await
            .expect("delivery failed");
    }
    println!("sent {} events", statuses.len());
}

src/bin/consumer.rs

Rust

use rdkafka::config::ClientConfig;
use rdkafka::consumer::{CommitMode, Consumer, StreamConsumer};
use rdkafka::Message;
use storefront_events::OrderEvent;

#[tokio::main]
async fn main() {
    let consumer: StreamConsumer = ClientConfig::new()
        .set("bootstrap.servers", "localhost:9092")
        .set("group.id", "shipping")
        .set("enable.auto.commit", "false")
        .set("auto.offset.reset", "earliest")
        .create()
        .expect("consumer");
    consumer.subscribe(&["orders"]).expect("subscribe");

    loop {
        let message = consumer.recv().await.expect("recv");
        if let Some(payload) = message.payload() {
            let event: OrderEvent = serde_json::from_slice(payload).expect("json");
            println!("order {} is {}", event.order_id, event.status);
        }
        consumer
            .commit_message(&message, CommitMode::Async)
            .expect("commit");
    }
}

src/bin/worker.rs

Rust

use rdkafka::config::ClientConfig;
use rdkafka::consumer::{CommitMode, Consumer, StreamConsumer};
use rdkafka::producer::{FutureProducer, FutureRecord};
use rdkafka::Message;
use sqlx::postgres::PgPoolOptions;
use sqlx::PgPool;
use std::time::Duration;
use storefront_events::OrderEvent;

async fn apply(pool: &PgPool, event: &OrderEvent) -> Result<bool, sqlx::Error> {
    let mut tx = pool.begin().await?;
    let inserted = sqlx::query(
        "INSERT INTO processed_events (event_id) VALUES ($1) ON CONFLICT DO NOTHING",
    )
    .bind(&event.event_id)
    .execute(&mut *tx)
    .await?
    .rows_affected();

    if inserted == 1 {
        sqlx::query("INSERT INTO shipments (order_id, status) VALUES ($1, $2)")
            .bind(event.order_id)
            .bind(&event.status)
            .execute(&mut *tx)
            .await?;
    }
    tx.commit().await?;
    Ok(inserted == 1)
}

#[tokio::main]
async fn main() {
    let pool = PgPoolOptions::new()
        .max_connections(5)
        .connect("postgres://postgres:postgres@localhost:5432/storefront")
        .await
        .expect("database");

    let consumer: StreamConsumer = ClientConfig::new()
        .set("bootstrap.servers", "localhost:9092")
        .set("group.id", "shipments-worker")
        .set("enable.auto.commit", "false")
        .set("auto.offset.reset", "earliest")
        .create()
        .expect("consumer");
    consumer.subscribe(&["orders"]).expect("subscribe");

    let producer: FutureProducer = ClientConfig::new()
        .set("bootstrap.servers", "localhost:9092")
        .set("acks", "all")
        .create()
        .expect("producer");

    loop {
        let message = consumer.recv().await.expect("recv");
        let payload = message.payload().unwrap_or_default();

        let mut handled = false;
        if let Ok(event) = serde_json::from_slice::<OrderEvent>(payload) {
            for attempt in 1..=3u64 {
                match apply(&pool, &event).await {
                    Ok(true) => println!("applied {}", event.event_id),
                    Ok(false) => println!("duplicate {}", event.event_id),
                    Err(error) => {
                        eprintln!("attempt {attempt} failed: {error}");
                        tokio::time::sleep(Duration::from_millis(200 * attempt)).await;
                        continue;
                    }
                }
                handled = true;
                break;
            }
        }

        if !handled {
            let record = FutureRecord::to("orders.dlq")
                .key(message.key().unwrap_or_default())
                .payload(payload);
            producer
                .send(record, Duration::from_secs(5))
                .await
                .expect("dead-letter send failed");
            println!("dead-lettered");
        }

        consumer
            .commit_message(&message, CommitMode::Sync)
            .expect("commit");
    }
}

Run, using a fresh Kafka topic and database (the docker run and docker exec lines from Project setup first):

Terminal

cargo run --bin producer
cargo run --bin worker        # terminal 2, keep running

Producer output, then worker output:

Output

sent 3 events
applied 7-0
applied 7-1
applied 7-2

Publish the same events again. The ids repeat, so the worker skips the effect:

Terminal

cargo run --bin producer

Output

sent 3 events
duplicate 7-0
duplicate 7-1
duplicate 7-2

The first line is the producer. The other three appear in the worker terminal.

Publish a message that is not JSON, then inspect the dead-letter topic and the tables:

Terminal

echo 'not json' | docker exec -i kafka /opt/kafka/bin/kafka-console-producer.sh \
  --topic orders --bootstrap-server localhost:9092
docker exec kafka /opt/kafka/bin/kafka-console-consumer.sh \
  --topic orders.dlq --from-beginning --max-messages 1 --bootstrap-server localhost:9092
docker exec pg psql -U postgres -d storefront -tA -c "SELECT order_id, status FROM shipments ORDER BY id"

Output

dead-lettered
not json
7|placed
7|paid
7|shipped

dead-lettered prints in the worker terminal. The other lines come from the two inspection commands. shipments holds three rows, not six, because the duplicates were skipped.

Summary

A queue deletes a message on acknowledgment. A log keeps messages, and each consumer group tracks an offset.
Kafka orders messages inside one partition only. The key selects the partition.
Consumer group parallelism is capped by the partition count.
Commit after handling gives at-least-once delivery. Commit before gives at-most-once.
Exactly-once effects come from idempotent handlers, such as a processed-events table in the same transaction as the effect.
Producer settings acks=all and enable.idempotence=true prevent loss and retry duplicates.
A message that cannot be processed goes to a dead-letter topic after bounded retries so it does not block its partition.
Consumer lag is the primary health signal. A rebalance pauses moved partitions.