Kafka #

Apache Kafka is a distributed event streaming platform designed to handle millions of events per second with low latency and high durability guarantees. Unlike traditional message queues like RabbitMQ that delete messages after consumption, Kafka stores events in a log that can be re-read — this enables event replay, audit trails, and multiple consumers each processing events independently. In Rust, the rdkafka crate is a binding to librdkafka — the battle-tested C library used in the largest production environments in the world. This article covers producers, consumers, serialization, transactions, and common architectural patterns used with Kafka.

Kafka Basic Concepts #

flowchart LR
    subgraph Producer
        P["Producer\n(Rust App)"]
    end

    subgraph Kafka Cluster
        T["Topic: pesanan\nPartition 0 | Partition 1 | Partition 2"]
        T --> P0["Partition 0\n[offset 0][offset 1][offset 2]"]
        T --> P1["Partition 1\n[offset 0][offset 1]"]
    end

    subgraph Consumer Group A
        C1["Consumer 1\n(processes Partition 0)"]
        C2["Consumer 2\n(processes Partition 1)"]
    end

    subgraph Consumer Group B
        C3["Consumer 3\n(processes all partitions)"]
    end

    P --> T
    P0 --> C1
    P1 --> C2
    T --> C3

Concept	Explanation
Topic	Event category — like a queue name
Partition	Parallel division within a topic
Offset	Unique position of each event in a partition
Producer	Sends events to a topic
Consumer	Reads events from a topic
Consumer Group	Several consumers sharing partition load
Broker	Kafka server — usually 3+ for production

Installation #

rdkafka requires librdkafka as a system dependency. The easiest option is to use the bundled CMake build:

[dependencies]
rdkafka = { version = "0.36", features = ["cmake-build"] }
tokio = { version = "1", features = ["full"] }
serde = { version = "1", features = ["derive"] }
serde_json = "1"
chrono = "0.4"

The cmake-build feature compiles librdkafka from source during cargo build — no extra system installation needed, but build times are longer.

Producer — Sending Events #

Basic Producer #

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

fn buat_producer(brokers: &str) -> FutureProducer {
    ClientConfig::new()
        .set("bootstrap.servers", brokers)
        .set("message.timeout.ms", "5000")
        // Delivery guarantee: "all" = wait for all replicas to acknowledge
        .set("acks", "all")
        // Automatic retry on failure
        .set("retries", "3")
        .set("retry.backoff.ms", "100")
        // Batching for high throughput
        .set("linger.ms", "5")         // wait 5ms for a larger batch
        .set("batch.size", "65536")    // max batch 64KB
        // Compression — saves bandwidth
        .set("compression.type", "snappy")
        .create()
        .expect("Failed to create producer")
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let producer = buat_producer("localhost:9092");

    // Send an event with a key and value
    let payload = serde_json::to_string(&serde_json::json!({
        "id": 1001,
        "aksi": "buat_pesanan",
        "total": 150_000
    }))?;

    let record = FutureRecord::to("pesanan")
        .key("user-42")          // the key determines the partition (consistent per user)
        .payload(&payload);

    match producer.send(record, Duration::from_secs(5)).await {
        Ok((partisi, offset)) => {
            println!("Event sent to partition {} offset {}", partisi, offset);
        }
        Err((e, _)) => {
            eprintln!("Failed to send: {}", e);
        }
    }

    Ok(())
}

Structured Producer with JSON Serialization #

use rdkafka::producer::{FutureProducer, FutureRecord};
use serde::{Deserialize, Serialize};
use std::time::Duration;

#[derive(Debug, Serialize, Deserialize, Clone)]
struct EventPesanan {
    pub id: u64,
    pub pengguna_id: u64,
    pub produk: Vec<ItemPesanan>,
    pub total: f64,
    pub status: String,
    pub dibuat_pada: String,
}

#[derive(Debug, Serialize, Deserialize, Clone)]
struct ItemPesanan {
    pub produk_id: u64,
    pub nama: String,
    pub jumlah: u32,
    pub harga: f64,
}

struct KafkaProducer {
    producer: FutureProducer,
    topic: String,
}

impl KafkaProducer {
    fn baru(brokers: &str, topic: &str) -> Self {
        KafkaProducer {
            producer: buat_producer(brokers),
            topic: topic.to_string(),
        }
    }

    async fn kirim<T: Serialize>(
        &self,
        key: &str,
        event: &T,
    ) -> Result<(i32, i64), rdkafka::error::KafkaError> {
        let payload = serde_json::to_string(event)
            .map_err(|e| rdkafka::error::KafkaError::MessageProduction(
                rdkafka::types::RDKafkaErrorCode::MessageSizeTooLarge
            ))?;

        let record = FutureRecord::to(&self.topic)
            .key(key)
            .payload(&payload);

        self.producer
            .send(record, Duration::from_secs(5))
            .await
            .map_err(|(e, _)| e)
    }

    async fn kirim_batch<T: Serialize>(
        &self,
        events: &[(String, T)],
    ) -> Vec<Result<(i32, i64), rdkafka::error::KafkaError>> {
        let mut hasil = Vec::new();
        for (key, event) in events {
            hasil.push(self.kirim(key, event).await);
        }
        hasil
    }
}

fn buat_producer(brokers: &str) -> FutureProducer {
    ClientConfig::new()
        .set("bootstrap.servers", brokers)
        .set("acks", "all")
        .set("retries", "3")
        .create()
        .expect("Failed to create producer")
}

Consumer — Reading Events #

Basic Consumer with a Consumer Group #

use rdkafka::config::ClientConfig;
use rdkafka::consumer::{Consumer, StreamConsumer};
use rdkafka::message::Message;
use futures::StreamExt;

fn buat_consumer(brokers: &str, group_id: &str) -> StreamConsumer {
    ClientConfig::new()
        .set("bootstrap.servers", brokers)
        .set("group.id", group_id)
        // Start from the beginning if no offset is stored yet
        .set("auto.offset.reset", "earliest")
        // Commit offsets manually (safer)
        .set("enable.auto.commit", "false")
        // Heartbeat interval to the broker
        .set("heartbeat.interval.ms", "3000")
        // Timeout if the consumer is silent too long
        .set("session.timeout.ms", "30000")
        .set("max.poll.interval.ms", "300000")
        .create()
        .expect("Failed to create consumer")
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let consumer: StreamConsumer = buat_consumer("localhost:9092", "grup-pemroses-pesanan");

    // Subscribe to one or many topics
    consumer.subscribe(&["pesanan", "pembayaran"])?;
    println!("Consumer active, waiting for events...");

    // Event stream — async iterator
    let mut stream = consumer.stream();

    while let Some(result) = stream.next().await {
        match result {
            Ok(message) => {
                let payload = message
                    .payload_view::<str>()
                    .unwrap_or(Ok(""))
                    .unwrap_or("");

                let key = message
                    .key_view::<str>()
                    .unwrap_or(Ok(""))
                    .unwrap_or("");

                println!(
                    "Topic: {}, Partition: {}, Offset: {}, Key: {}, Payload: {}",
                    message.topic(),
                    message.partition(),
                    message.offset(),
                    key,
                    &payload[..payload.len().min(100)]
                );

                // Process the event
                if let Ok(event) = serde_json::from_str::<serde_json::Value>(payload) {
                    proses_event(&event).await;
                }

                // Commit the offset after successful processing
                consumer.commit_message(&message, rdkafka::consumer::CommitMode::Async)?;
            }
            Err(e) => {
                eprintln!("Kafka error: {}", e);
            }
        }
    }

    Ok(())
}

async fn proses_event(event: &serde_json::Value) {
    println!("Processing: {:?}", event["aksi"]);
    // ... business logic
}

Consumer with Typed Deserialization #

use rdkafka::consumer::{Consumer, StreamConsumer};
use rdkafka::message::Message;
use futures::StreamExt;

struct EventConsumer {
    consumer: StreamConsumer,
}

impl EventConsumer {
    fn baru(brokers: &str, group_id: &str, topics: &[&str]) -> Self {
        let consumer: StreamConsumer = buat_consumer(brokers, group_id);
        consumer.subscribe(topics).expect("Failed to subscribe");
        EventConsumer { consumer }
    }

    async fn proses_loop<T, F, Fut>(&self, handler: F)
    where
        T: for<'de> serde::Deserialize<'de>,
        F: Fn(T, i32, i64) -> Fut,
        Fut: std::future::Future<Output = Result<(), String>>,
    {
        let mut stream = self.consumer.stream();

        while let Some(result) = stream.next().await {
            match result {
                Ok(message) => {
                    let partisi = message.partition();
                    let offset = message.offset();

                    let payload = match message.payload_view::<str>() {
                        Some(Ok(s)) => s,
                        _ => {
                            eprintln!("Payload is not UTF-8 at offset {}", offset);
                            continue;
                        }
                    };

                    match serde_json::from_str::<T>(payload) {
                        Ok(event) => {
                            match handler(event, partisi, offset).await {
                                Ok(_) => {
                                    // Commit after success
                                    let _ = self.consumer.commit_message(
                                        &message,
                                        rdkafka::consumer::CommitMode::Async,
                                    );
                                }
                                Err(e) => {
                                    eprintln!("Handler error at offset {}: {}", offset, e);
                                    // Could implement a dead letter queue here
                                }
                            }
                        }
                        Err(e) => {
                            eprintln!("Failed to deserialize at offset {}: {}", offset, e);
                        }
                    }
                }
                Err(e) => eprintln!("Kafka error: {}", e),
            }
        }
    }
}

fn buat_consumer(brokers: &str, group_id: &str) -> StreamConsumer {
    ClientConfig::new()
        .set("bootstrap.servers", brokers)
        .set("group.id", group_id)
        .set("auto.offset.reset", "earliest")
        .set("enable.auto.commit", "false")
        .create()
        .expect("Failed to create consumer")
}

Topic Management with the Admin API #

use rdkafka::admin::{AdminClient, AdminOptions, NewTopic, TopicReplication};
use rdkafka::client::DefaultClientContext;
use rdkafka::config::ClientConfig;

fn buat_admin(brokers: &str) -> AdminClient<DefaultClientContext> {
    ClientConfig::new()
        .set("bootstrap.servers", brokers)
        .create()
        .expect("Failed to create admin client")
}

async fn buat_topic(
    admin: &AdminClient<DefaultClientContext>,
    nama: &str,
    partisi: i32,
    replikasi: i32,
) -> Result<(), Box<dyn std::error::Error>> {
    let topic = NewTopic::new(
        nama,
        partisi,
        TopicReplication::Fixed(replikasi),
    )
    // 7-day retention
    .set("retention.ms", "604800000")
    // Broker-side compression
    .set("compression.type", "snappy");

    let opsi = AdminOptions::new()
        .operation_timeout(Some(std::time::Duration::from_secs(30)));

    let hasil = admin.create_topics(&[topic], &opsi).await?;

    for r in hasil {
        match r {
            Ok(nama) => println!("Topic '{}' created successfully", nama),
            Err((nama, e)) => {
                // An existing topic isn't an error in production
                if e == rdkafka::types::RDKafkaErrorCode::TopicAlreadyExists {
                    println!("Topic '{}' already exists", nama);
                } else {
                    eprintln!("Failed to create topic '{}': {:?}", nama, e);
                }
            }
        }
    }

    Ok(())
}

async fn hapus_topic(
    admin: &AdminClient<DefaultClientContext>,
    nama: &str,
) -> Result<(), Box<dyn std::error::Error>> {
    let opsi = AdminOptions::new();
    let hasil = admin.delete_topics(&[nama], &opsi).await?;
    for r in hasil {
        match r {
            Ok(nama) => println!("Topic '{}' deleted", nama),
            Err((nama, e)) => eprintln!("Failed to delete '{}': {:?}", nama, e),
        }
    }
    Ok(())
}

Kafka Transactions #

Kafka transactions ensure a group of events is sent atomically — all succeed or all are aborted. Important for “exactly-once semantics” patterns:

use rdkafka::producer::{BaseRecord, ThreadedProducer};

fn buat_producer_transaksional(brokers: &str, transactional_id: &str) -> ThreadedProducer<rdkafka::producer::DefaultProducerContext> {
    ClientConfig::new()
        .set("bootstrap.servers", brokers)
        .set("transactional.id", transactional_id)
        .set("acks", "all")
        .set("enable.idempotence", "true")
        .create()
        .expect("Failed to create transactional producer")
}

fn kirim_dengan_transaksi(
    producer: &ThreadedProducer<rdkafka::producer::DefaultProducerContext>,
    events: &[(&str, &str, &str)],  // (topic, key, payload)
) -> Result<(), rdkafka::error::KafkaError> {
    // Initialize transactions (once at startup)
    producer.init_transactions(std::time::Duration::from_secs(10))?;

    // Begin the transaction
    producer.begin_transaction()?;

    for (topic, key, payload) in events {
        producer.send(
            BaseRecord::to(topic)
                .key(*key)
                .payload(*payload),
        ).map_err(|(e, _)| e)?;
    }

    // Commit all events at once
    producer.commit_transaction(std::time::Duration::from_secs(10))?;
    println!("Transaction succeeded: {} events sent", events.len());
    Ok(())
}

Event-Driven Architecture Patterns #

Pattern: Outbox for Consistency #

A pattern that ensures events are only sent to Kafka if the database change was successfully stored:

sequenceDiagram
    participant API
    participant DB as Database
    participant Outbox as Outbox Table
    participant Relay as Outbox Relay
    participant Kafka

    API->>DB: BEGIN TRANSACTION
    API->>DB: INSERT pesanan
    API->>Outbox: INSERT event (in the same transaction)
    DB->>API: COMMIT

    Relay->>Outbox: Poll unsent events
    Relay->>Kafka: Produce event
    Kafka->>Relay: Acknowledged
    Relay->>Outbox: Mark as sent

use rdkafka::producer::FutureProducer;
use serde::{Deserialize, Serialize};

// The outbox table in the database
#[derive(Debug, Serialize, Deserialize)]
struct OutboxEvent {
    pub id: i64,
    pub topic: String,
    pub key: String,
    pub payload: String,
    pub terkirim: bool,
    pub dibuat_pada: chrono::DateTime<chrono::Utc>,
}

// Relay: read from outbox, send to Kafka, mark as sent
async fn outbox_relay(
    pool: sqlx::PgPool,
    producer: FutureProducer,
) {
    loop {
        // Fetch unsent events
        let events = sqlx::query_as!(
            OutboxEvent,
            "SELECT id, topic, key, payload, terkirim, dibuat_pada
             FROM outbox
             WHERE terkirim = FALSE
             ORDER BY id ASC
             LIMIT 100"
        )
        .fetch_all(&pool)
        .await
        .unwrap_or_default();

        for event in &events {
            let record = FutureRecord::to(&event.topic)
                .key(&event.key)
                .payload(&event.payload);

            match producer.send(record, std::time::Duration::from_secs(5)).await {
                Ok(_) => {
                    // Mark as sent
                    let _ = sqlx::query!(
                        "UPDATE outbox SET terkirim = TRUE WHERE id = $1",
                        event.id
                    )
                    .execute(&pool)
                    .await;
                }
                Err((e, _)) => {
                    eprintln!("Failed to send event {}: {}", event.id, e);
                }
            }
        }

        // Wait before the next poll
        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
    }
}

Consumer with a Dead Letter Queue #

If event processing fails repeatedly, send it to a DLQ (Dead Letter Queue) for investigation:

use rdkafka::consumer::{Consumer, StreamConsumer};
use rdkafka::message::Message;
use rdkafka::producer::{FutureProducer, FutureRecord};
use futures::StreamExt;

struct ConsumerDenganDLQ {
    consumer: StreamConsumer,
    producer: FutureProducer,
    dlq_topic: String,
    maks_retry: u32,
}

impl ConsumerDenganDLQ {
    async fn jalankan<F, Fut>(&self, handler: F)
    where
        F: Fn(String) -> Fut,
        Fut: std::future::Future<Output = Result<(), String>>,
    {
        let mut stream = self.consumer.stream();

        while let Some(Ok(message)) = stream.next().await {
            let payload = message
                .payload_view::<str>()
                .unwrap_or(Ok(""))
                .unwrap_or("")
                .to_string();

            let mut berhasil = false;
            let mut percobaan = 0;

            // Retry up to maks_retry times
            while percobaan < self.maks_retry {
                match handler(payload.clone()).await {
                    Ok(_) => {
                        berhasil = true;
                        break;
                    }
                    Err(e) => {
                        percobaan += 1;
                        eprintln!("Attempt {}/{}: {}", percobaan, self.maks_retry, e);
                        tokio::time::sleep(
                            std::time::Duration::from_millis(100 * percobaan as u64)
                        ).await;
                    }
                }
            }

            if !berhasil {
                // Send to the DLQ
                let dlq_payload = serde_json::json!({
                    "payload_asli": payload,
                    "error": "Exceeded max retries",
                    "topic_asal": message.topic(),
                    "partisi": message.partition(),
                    "offset": message.offset(),
                    "gagal_pada": chrono::Utc::now().to_rfc3339()
                }).to_string();

                let _ = self.producer.send(
                    FutureRecord::to(&self.dlq_topic)
                        .payload(&dlq_payload),
                    std::time::Duration::from_secs(5),
                ).await;

                eprintln!("Event sent to DLQ: {}", &self.dlq_topic);
            }

            // Always commit — even those sent to the DLQ
            let _ = self.consumer.commit_message(
                &message,
                rdkafka::consumer::CommitMode::Async,
            );
        }
    }
}

Summary #

acks = "all" for delivery guarantees — the producer waits for all broker replicas to acknowledge before considering it successful. Slower but no data loss.
The key determines the partition — events with the same key always go to the same partition, guaranteeing order per key. Use the user/entity ID as the key.
enable.auto.commit = false — always commit manually after the event is successfully processed. Auto-commit is risky: events can be committed before processing if the app crashes.
Consumer groups for horizontal scaling — add new consumer instances in the same group to share the load. The max number of active consumers equals the number of partitions.
auto.offset.reset = "earliest" — start from the oldest event if no offset is stored. Use "latest" if you only need new events.
The outbox pattern for consistency — store events in an outbox table in the same database transaction as the data change, then relay to Kafka. Guarantees no events are lost when the app crashes.
Dead Letter Queues for failed events — instead of blocking the consumer when processing fails, send to the DLQ for investigation and manual retry.
Kafka transactions for exactly-once — use transactional.id and enable.idempotence to guarantee events aren’t duplicated even with network retries.
Enough partitions per topic — more partitions = more parallelism, but also more overhead. Start with 3–6 partitions per topic for most cases.

← Previous: Elasticsearch Next: RabbitMQ →