Chapter 8 — Streaming and Event-Driven Architecture in Rust

Opening — When the Data Arrived Too Late

The system hadn’t failed. That was the hardest part to explain. All jobs had completed. Logs were green. Metrics dashboards showed no errors. And yet, the decision made at 9:30 AM was the wrong one. Not because the data was missing — but because it was late. It was a click made by a user at 8:43. A purchase that wasn’t processed in time. An alert that never fired because the ETL job was still running. We didn’t lose money. We lost timing. We lost truth. And in that moment, we realized: the warehouse wasn’t the problem. The assumption was.

“That’s when we realized: the batch wasn’t the problem. The assumption was.”

8.1 Kafka — The Legacy King

You don’t need another broker. You need a new latency contract. Kafka brought structure where there was none. It replaced queues with replayable logs. It gave us partitions, durability, replication. But Kafka never promised “real time”. Kafka is a log, not a reaction. And somewhere along the way, we started asking it to be both.

Scene

We were using Kafka with Python consumers. And it worked — until volume hit 250,000 events per minute. Then it started lying. Messages were silently duplicated. Commits were skipped. Consumer groups were rebalancing constantly. And our dashboards drifted without anyone noticing. The issue wasn’t the broker. It was the runtime.

Why Rust Matters Here

We rewrote the consumer using `rust-rdkafka`. What changed wasn’t just performance — it was clarity.

use rdkafka::consumer::{Consumer, StreamConsumer, CommitMode};
use rdkafka::ClientConfig;

let consumer: StreamConsumer = ClientConfig::new()
    .set("bootstrap.servers", "localhost:9092")
    .set("group.id", "realtime-analytics")
    .create()
    .expect("Consumer creation failed");

let stream = consumer.start();
stream.for_each(|message| async {
    match message {
        Ok(m) => {
            let payload = m.payload().unwrap_or_default();
            process_event(payload).await;
            consumer.commit_message(&m, CommitMode::Async).unwrap();
        }
        Err(e) => eprintln!("Kafka error: {}", e),
    }
}).await;

We moved from polling loops and implicit timeouts to a fully asynchronous stream. And with that came control: over commit strategies, rebalancing callbacks, buffer size, and the entire flow contract.

Insight

• When to use it: When Kafka is solid but your consumption strategy isn’t.
• What it replaces: Batch-like consumers pretending to be streaming systems.
• What it breaks: The illusion that your language doesn’t matter.
• What it doesn’t replace: You still need to define offset strategies and backpressure logic.

Design Question: If latency defines trust, why are you letting the slowest runtime sign the contract?

8.2 Pulsar — The Architecture Kafka Never Promised

You’re not paying for Kafka. You’re paying for what it expects. Apache Pulsar didn’t try to replace Kafka — it rewrote the model entirely. In Pulsar, brokers don’t store data. Storage is delegated to BookKeeper. This means brokers are lightweight, topics are scalable by design, and multi-tenancy is a feature, not a tax.

Scene

A team wanted to open 800 topics — one per customer. Kafka imploded: Zookeeper became a bottleneck, rebalance storms began every 90 seconds, and metrics were skewed. We migrated to Pulsar. Nothing broke. Because it was built for this from the start.

Rust Client (pulsar-rs)

use pulsar::{Pulsar, TokioExecutor, SubType};

let pulsar: Pulsar<_> = Pulsar::builder("pulsar://localhost:6650", TokioExecutor)
    .build()
    .await?;

let mut consumer = pulsar
    .consumer()
    .with_topic("persistent://public/default/events")
    .with_subscription_type(SubType::Shared)
    .with_subscription("analytics-sub")
    .build()
    .await?;

while let Some(msg) = consumer.try_next().await? {
    let data = msg.deserialize()?;
    handle(data).await;
    consumer.ack(&msg).await?;
}

It’s idiomatic, typed, async. And it doesn’t require a JVM, a registry, or a prayer.

Insight

• When to use it: When you have many topics, regions, or teams sharing infrastructure.
• What it replaces: Kafka’s operational tax on scale.
• What it breaks: The idea that storage and dispatch must live on the same node.
• What it doesn’t replace: Kafka’s ecosystem of connectors — you’ll need to build more yourself.

Design Question: If you weren’t afraid of opening 1,000 topics, how would your architecture change?

8.3 Fluvio — The Linux of Event Streams

You don’t need a broker. You need a stream that speaks your language. Fluvio is not Kafka reimagined. It’s streaming from scratch, written in Rust. Lightweight. Embeddable. Composable. It runs without the JVM, without Zookeeper, and without overhead.

Scene

We needed a streaming platform that could live on the edge — Low resources, low latency, no central coordination. Kafka was overkill. Pulsar was too heavy. Fluvio launched in seconds. And it worked.

Architecture

• SC (Streaming Controller): Controls and orchestrates
• SPUs (Streaming Processing Units): Handle ingestion and storage
• SmartModules: WASM modules that transform events inside the stream

Producing and Consuming in Rust

use fluvio::{Fluvio, RecordKey};

let fluvio = Fluvio::connect().await?;
let producer = fluvio.topic_producer("logs").await?;
producer.send(RecordKey::NULL, "event:login").await?;

let consumer = fluvio.partition_consumer("logs", 0).await?;
let mut stream = consumer.stream(None).await?;

while let Some(Ok(record)) = stream.next().await {
    println!(">> {:?}", String::from_utf8_lossy(record.value()));
}

Insight

• When to use it: When you control the infrastructure and need minimal runtime.
• What it replaces: Kafka clusters in constrained or embedded environments.
• What it breaks: The idea that processing must be external.
• What it doesn’t replace: Flink-level topologies — Fluvio is lightweight, not omniscient.

Design Question: What would your system look like if every stream could run its own logic — in place?

8.4 SmartModules — When Code Travels With the Event

You’re not processing the data. The data is carrying the processing. SmartModules are small Rust programs compiled to WebAssembly. They run inside Fluvio. That means the transformation happens before the event even leaves the stream.

Scene

We had logs coming in from hundreds of devices. Most of them were noise. We used to filter them downstream, wasting bandwidth, memory, and cycles. With SmartModules, we filtered them before they left the SPU.

#[smartmodule(filter)]
pub fn my_filter(record: &Record) -> bool {
    let value = std::str::from_utf8(record.value.as_ref()).unwrap_or("");
    value.contains("ERROR")
}

This compiled into WASM. Loaded into Fluvio. And filtered every message on the wire.

Insight

• When to use it: When downstream isn’t an option — or a luxury.
• What it replaces: Microservices that do 10 lines of transformation.
• What it breaks: The illusion that routing logic needs an external process.
• What it doesn’t replace: Stateful stream processing across topics or complex joins.

Design Question: If you could inject logic into the stream itself, would you still need half your services?

8.5 Materialize & RisingWave — SQL Without Time Lag

It’s not SQL that changed. It’s the time model behind it. Materialize and RisingWave let you write SQL on top of streams — And the results update automatically. No cron. No “refresh every 5 minutes”. No nightly batch. Just live views.

Scene

We were running hourly jobs to aggregate events into a warehouse. A “simple” aggregation of product sales by region. Until we replaced it with:

CREATE MATERIALIZED VIEW sales_by_region AS
SELECT region, SUM(amount)
FROM purchases_stream
GROUP BY region;

And never ran the job again.

Comparison

Insight

• When to use it: When you need real-time analytical queries with SQL.
• What it replaces: Scheduled jobs, refresh cycles, and stale dashboards.
• What it breaks: The idea that “views must be rebuilt”.
• What it doesn’t replace: Warehouses for historical analytics over large windows.

Design Question: If your queries could evolve with the data, what would you stop precomputing?

8.6 Vector — The Nervous System Begins Here

Observability isn’t a dashboard. It’s what you do when the logs arrive in time. Vector, written in Rust, is a telemetry agent. But more than that — it’s a real-time log to stream bridge. It reads from syslog, journald, files, metrics. Transforms them. And sends them to Kafka, Pulsar, ClickHouse — instantly.

Scene

Our alerts were delayed because logs took 10 minutes to hit Elasticsearch. With Vector, logs were filtered, structured, and routed within seconds.

[sources.syslog]
type = "syslog"

[transforms.filter_errors]
type = "filter"
inputs = ["syslog"]
condition = '.message.contains("ERROR")'

[sinks.kafka]
type = "kafka"
inputs = ["filter_errors"]
bootstrap_servers = "localhost:9092"
topic = "system-logs"

This wasn’t observability as insight. It was observability as reaction.

Insight

• When to use it: When logs should become events, not documents.
• What it replaces: ELK pipelines for reactive contexts.
• What it breaks: The passive model of log collection.
• What it doesn’t replace: Long-term retention, complex log queries.

Design Question: If logs were part of your pipeline, not after it — would you still treat them as evidence, or as signals?

8.7 End-to-End Real-Time ETL: Kafka + Rust + Warehouse

This is not a diagram. This is a battle map. We built a real-time ingestion pipeline: Kafka as the ingestion bus, Rust as the transformer, and Snowflake as the sink. And we did it without CSVs. Without cron. Without staging tables.

Architecture Flow

[Web Events]
↓
[Kafka Topic: user_actions]
↓
[Rust Service: stream_processor.rs]
- Deserialize
- Enrich with geo IP
- Format as Arrow batch
↓
[Snowflake: Snowpipe Streaming API]
↓
[Materialized View: actions_by_region]

Rust + Arrow + Snowflake

use arrow2::{array::*, datatypes::*, record_batch::*, io::ipc::write::write_file};
use std::fs::File;

let schema = Schema::from(vec![
    Field::new("user_id", DataType::Utf8, false),
    Field::new("country", DataType::Utf8, false),
]);

let batch = RecordBatch::try_new(
    Arc::new(schema),
    vec![
        Arc::new(Utf8Array::<i32>::from_slice(["abc", "def"])),
        Arc::new(Utf8Array::<i32>::from_slice(["US", "UK"])),
    ],
)?;

let mut file = File::create("batch.arrow")?;
write_file(&mut file, &batch, &WriteOptions::default())?;

Final Tradeoffs

Insight

• When to use it: When freshness defines your business.
• What it replaces: Every pipeline where “minutes” is the smallest unit.
• What it breaks: The boundary between ingestion and analytics.
• What it doesn’t replace: Batch jobs for historical backfills.

Design Question: If freshness was guaranteed, how would your data strategy change tomorrow?