Implementation:Fede1024 Rust rdkafka Roundtrip Latency Example
| Knowledge Sources | |
|---|---|
| Domains | Benchmarking, Performance, Latency_Measurement |
| Last Updated | 2026-02-07 19:00 GMT |
Overview
Example application that measures end-to-end produce-consume latency using a Kafka topic, recording results in an HDR histogram and reporting percentile statistics.
Description
The roundtrip example creates a FutureProducer and a StreamConsumer connected to the same topic. A background tokio task continuously produces messages stamped with wall-clock time in milliseconds. The main task consumes messages in three phases: a 10-second warm-up (discards measurements), a 10-second recording period (computes delta between send and receive timestamps, records in an hdrhistogram::Histogram), then stops. Reports mean, p50, p90, and p99 latency statistics.
Usage
Run this example to benchmark Kafka roundtrip latency in your environment. Useful for evaluating broker performance, network latency, and the overhead of rust-rdkafka's async pipeline.
Code Reference
Source Location
- Repository: Fede1024_Rust_rdkafka
- File: examples/roundtrip.rs
- Lines: 1-106
Signature
#[tokio::main]
async fn main() // CLI entry point with clap argument parsing
fn now() -> i64 // Returns current time in milliseconds since UNIX_EPOCH
Import
use rdkafka::producer::FutureProducer;
use rdkafka::consumer::{StreamConsumer, Consumer};
use rdkafka::config::ClientConfig;
use rdkafka::Message;
use hdrhistogram::Histogram;
I/O Contract
Inputs
| Name | Type | Required | Description |
|---|---|---|---|
| brokers | String (CLI arg) | Yes | Comma-separated broker list |
| topic | String (CLI arg) | Yes | Topic for roundtrip measurement |
Outputs
| Name | Type | Description |
|---|---|---|
| stdout | Text | Latency statistics: mean, p50, p90, p99 in milliseconds |
Usage Examples
Running the Benchmark
# Measure roundtrip latency on a topic
cargo run --example roundtrip -- --brokers localhost:9092 --topic latency-test
Key Pattern: Timestamp-Based Latency Measurement
use std::time::{SystemTime, UNIX_EPOCH};
fn now() -> i64 {
let start = SystemTime::now();
let since_epoch = start.duration_since(UNIX_EPOCH).unwrap();
since_epoch.as_millis() as i64
}
// In producer: embed timestamp as message payload
let payload = format!("{}", now());
// In consumer: compute latency
let send_time: i64 = std::str::from_utf8(msg.payload().unwrap())
.unwrap()
.parse()
.unwrap();
let latency = now() - send_time;
histogram.record(latency as u64).unwrap();