Sensor pipeline

Configure the broker

Open a broker with LZ4 compression and a retention policy of 500 records per partition. The BrokerConfig struct uses builder-style defaults—..BrokerConfig::new(path) fills in auto_create_topics: true and default_partitions: 1.

let config = BrokerConfig {
    compression: Compression::Lz4,
    default_retention: RetentionConfig {
        max_records: Some(500),
    },
    ..BrokerConfig::new(&data_path)
};
let broker = Broker::open(config).unwrap();

Produce individual records

Send 200 sensor readings one at a time. Each ProducerRecord carries a topic name, an optional key, and a string value. The key is the sensor ID—merkql hashes the key to determine partition assignment, so all readings from sensor-a go to the same partition.

let producer = Broker::producer(&broker);
let sensors = ["sensor-a", "sensor-b", "sensor-c", "sensor-d"];

for i in 0..200 {
    let sensor_id = sensors[i % sensors.len()];
    let reading = SensorReading {
        temp: 20.0 + (i as f64 * 0.1) % 20.0,
        ts: 1700000000 + i as u64,
    };
    let value = serde_json::to_string(&reading).unwrap();
    producer.send(&ProducerRecord::new(
        "sensors",
        Some(sensor_id.to_string()),
        value,
    )).unwrap();
}

Each send() call appends the record to the partition, updates the merkle tree, flushes the index, and fsyncs. This guarantees durability per record but costs ~31µs each.

Batch produce

Send 800 more readings in 8 batches of 100 using send_batch(). Internally, the batch API buffers all writes and performs a single fsync at the end—significantly higher throughput for bulk ingestion.

for batch_num in 0..8 {
    let batch: Vec<ProducerRecord> = (0..100)
        .map(|j| {
            let idx = 200 + batch_num * 100 + j;
            ProducerRecord::new(
                "sensors",
                Some(sensors[idx % sensors.len()].to_string()),
                serde_json::to_string(&reading).unwrap(),
            )
        })
        .collect();
    producer.send_batch(&batch).unwrap();
}

Consume with the “analytics” group

Create a consumer in the “analytics” group with auto_commit: false and OffsetReset::Earliest. Subscribe to the “sensors” topic and poll. Because retention is set to 500 records, only the most recent 500 are available—the first 500 have been logically trimmed.

let mut consumer = Broker::consumer(
    &broker,
    ConsumerConfig {
        group_id: "analytics".into(),
        auto_commit: false,
        offset_reset: OffsetReset::Earliest,
    },
);
consumer.subscribe(&["sensors"]).unwrap();
let records = consumer.poll(Duration::from_millis(100)).unwrap();

After processing, commit_sync() persists the current offset to disk so the group can resume later without re-reading committed records.

// Compute min/max/avg per sensor, then commit
consumer.commit_sync().unwrap();
consumer.close().unwrap();

Independent consumer group

A second consumer group (“alerts”) reads the same topic independently. It filters for readings above 35.0°C and prints alerts. Each group has its own committed offsets—the “analytics” group's earlier commit has no effect on where “alerts” starts reading.

let mut consumer = Broker::consumer(
    &broker,
    ConsumerConfig {
        group_id: "alerts".into(),
        auto_commit: false,
        offset_reset: OffsetReset::Earliest,
    },
);
consumer.subscribe(&["sensors"]).unwrap();
let records = consumer.poll(Duration::from_millis(100)).unwrap();

for record in &records {
    let reading: SensorReading = serde_json::from_str(&record.value).unwrap();
    if reading.temp > 35.0 {
        println!("ALERT: {} temp={:.1}",
            record.key.as_deref().unwrap_or("unknown"), reading.temp);
    }
}

Broker restart

Drop the broker entirely and reopen it from the same directory. All state—topics, partitions, merkle trees, consumer group offsets—is recovered from disk. This simulates a process restart.

drop(broker);

let config = BrokerConfig {
    compression: Compression::Lz4,
    default_retention: RetentionConfig {
        max_records: Some(500),
    },
    ..BrokerConfig::new(&data_path)
};
let broker = Broker::open(config).unwrap();

Resume without duplicates

The “analytics” consumer reconnects after the restart. Because it committed its offsets before the broker was dropped, it resumes from where it left off and receives zero records.

let mut consumer = Broker::consumer(&broker, analytics_config);
consumer.subscribe(&["sensors"]).unwrap();
let records = consumer.poll(Duration::from_millis(100)).unwrap();
// records.len() == 0

Incremental consumption

Produce 100 more readings, then poll with the “analytics” group again. It receives exactly the 100 new records—no more, no less. This confirms that offset tracking works correctly across broker restarts.

// Produce 100 more readings...

let records = consumer.poll(Duration::from_millis(100)).unwrap();
assert_eq!(records.len(), 100);

Disk usage

The example prints the total size of the data directory. With LZ4 compression and 1,100 JSON sensor readings, the entire dataset (including merkle tree nodes, indices, and consumer group state) fits in approximately 563 KB.