use std::{
collections::{BTreeSet, HashMap},
env, fs,
path::PathBuf,
process::Command,
sync::{
Arc, Mutex,
atomic::{AtomicBool, AtomicU8, AtomicU64, Ordering},
},
thread,
time::{Duration, Instant, SystemTime, UNIX_EPOCH},
};
use bytes::Bytes;
use datum::{Keep, Sink, Source, StreamCompletion, StreamError};
use datum_mq::profile::{self, ProfileBucket, ProfileBucketSnapshot, ProfileSnapshot};
use datum_mq::{
KafkaConsumerBackend, KafkaConsumerSettings, KafkaControl, KafkaPayloadBatch,
KafkaProducerBackend, KafkaProducerSettings, KafkaSink, KafkaSource, ProducerRecord,
Subscription,
};
use futures::executor::block_on;
use rdkafka::{
ClientConfig,
admin::{AdminClient, AdminOptions, NewTopic, TopicReplication},
client::DefaultClientContext,
consumer::{BaseConsumer, CommitMode, Consumer},
producer::{BaseProducer, BaseRecord, Producer},
topic_partition_list::{Offset, TopicPartitionList},
types::RDKafkaErrorCode,
util::Timeout,
};
const HEADER_BYTES: usize = 24;
fn main() {
let config = BenchConfig::from_args(env::args().skip(1));
profile::set_enabled(env_flag("DATUM_MQ_PROFILE"));
profile::set_native_enabled(env_flag("DATUM_KAFKA_NATIVE_PROFILE"));
run_trials(config);
}
fn env_flag(name: &str) -> bool {
env::var(name)
.ok()
.is_some_and(|value| !matches!(value.as_str(), "" | "0" | "false" | "FALSE" | "no" | "NO"))
}
#[derive(Debug, Clone)]
struct BenchConfig {
scenario: String,
bootstrap: String,
records: usize,
payload_bytes: usize,
partitions: i32,
warmups: usize,
iterations: usize,
out: Option<PathBuf>,
commit_batch: usize,
client_id_prefix: String,
consumer_backend: KafkaConsumerBackend,
producer_backend: KafkaProducerBackend,
producer_max_in_flight: usize,
producer_enable_idempotence: Option<bool>,
compression: String,
}
impl BenchConfig {
fn from_args<I>(args: I) -> Self
where
I: IntoIterator<Item = String>,
{
let args = parse_args(args);
let native_compare = native_compare_binary();
let default_scenario = if native_compare {
"consume_committable_1M"
} else {
"produce_1M_256B"
};
let default_client_prefix = if native_compare {
"datum-mq-native"
} else {
"datum-mq-datum"
};
Self {
scenario: args
.get("scenario")
.cloned()
.unwrap_or_else(|| default_scenario.to_owned()),
bootstrap: args
.get("bootstrap")
.cloned()
.or_else(|| env::var("MQ_BOOTSTRAP_SERVERS").ok())
.unwrap_or_else(|| "127.0.0.1:9092".to_owned()),
records: parse_usize(&args, "records", 1_000_000),
payload_bytes: parse_usize(&args, "payload-bytes", 256).max(HEADER_BYTES),
partitions: parse_i32(&args, "partitions", 16).max(1),
warmups: parse_usize(&args, "warmups", 2),
iterations: parse_usize(&args, "iterations", 5).max(1),
out: args.get("out").map(PathBuf::from),
commit_batch: parse_usize(&args, "commit-batch", 10_000).max(1),
client_id_prefix: args
.get("client-id-prefix")
.cloned()
.unwrap_or_else(|| default_client_prefix.to_owned()),
consumer_backend: args.get("consumer-backend").map_or_else(
|| {
if native_compare {
KafkaConsumerBackend::Native
} else {
KafkaConsumerBackend::default()
}
},
|backend| parse_consumer_backend(backend),
),
producer_backend: args.get("producer-backend").map_or_else(
|| {
if native_compare {
KafkaProducerBackend::Native
} else {
KafkaProducerBackend::default()
}
},
|backend| parse_producer_backend(backend),
),
producer_max_in_flight: parse_usize(&args, "producer-max-in-flight", 1).max(1),
producer_enable_idempotence: args
.get("producer-enable-idempotence")
.map(|_| parse_bool(&args, "producer-enable-idempotence", false)),
compression: parse_compression(args.get("compression").map(String::as_str)),
}
}
fn client_name(&self) -> &'static str {
if self.scenario == "produce_1M_256B" {
return match self.producer_backend {
KafkaProducerBackend::Rdkafka => "datum",
KafkaProducerBackend::Native => "datum-mq-native",
};
}
match self.consumer_backend {
KafkaConsumerBackend::Rdkafka => "datum",
KafkaConsumerBackend::Native => "datum-mq-native",
}
}
}
fn parse_compression(value: Option<&str>) -> String {
let value = value.unwrap_or("none").trim().to_ascii_lowercase();
match value.as_str() {
"none" | "lz4" | "zstd" => value,
other => panic!("unknown compression '{other}'; expected none, lz4, or zstd"),
}
}
fn parse_producer_backend(value: &str) -> KafkaProducerBackend {
match value {
"rdkafka" | "Rdkafka" => KafkaProducerBackend::Rdkafka,
"native" | "Native" => KafkaProducerBackend::Native,
other => panic!("unknown producer backend '{other}'; expected rdkafka or native"),
}
}
fn native_compare_binary() -> bool {
env::args()
.next()
.is_some_and(|binary| binary.contains("native_mq_compare"))
}
fn parse_consumer_backend(value: &str) -> KafkaConsumerBackend {
match value {
"rdkafka" | "Rdkafka" => KafkaConsumerBackend::Rdkafka,
"native" | "Native" => KafkaConsumerBackend::Native,
other => panic!("unknown consumer backend '{other}'; expected rdkafka or native"),
}
}
fn parse_args<I>(args: I) -> HashMap<String, String>
where
I: IntoIterator<Item = String>,
{
let mut out = HashMap::new();
let mut args = args.into_iter().peekable();
while let Some(raw) = args.next() {
let Some(trimmed) = raw.strip_prefix("--") else {
panic!("unexpected argument '{raw}'");
};
if let Some((key, value)) = trimmed.split_once('=') {
out.insert(key.to_owned(), value.to_owned());
} else if args.peek().is_some_and(|next| !next.starts_with("--")) {
let value = args.next().expect("peeked value exists");
out.insert(trimmed.to_owned(), value);
} else {
out.insert(trimmed.to_owned(), "true".to_owned());
}
}
out
}
fn parse_usize(args: &HashMap<String, String>, key: &str, default: usize) -> usize {
args.get(key)
.and_then(|value| value.parse().ok())
.unwrap_or(default)
}
fn parse_bool(args: &HashMap<String, String>, key: &str, default: bool) -> bool {
args.get(key).map_or(default, |value| {
matches!(
value.trim().to_ascii_lowercase().as_str(),
"1" | "true" | "yes"
)
})
}
fn parse_i32(args: &HashMap<String, String>, key: &str, default: i32) -> i32 {
args.get(key)
.and_then(|value| value.parse().ok())
.unwrap_or(default)
}
fn run_trials(config: BenchConfig) {
validate_scenario(&config.scenario);
let mut rows = Vec::new();
for trial in 0..config.warmups {
let topic = topic_name(&config.scenario, &format!("warmup-{trial}"));
create_topic(&config.bootstrap, &topic, config.partitions);
let prepared_group = prepare_trial(&config, &topic, "warmup", trial);
if profile::enabled() {
profile::reset();
}
if profile::native_enabled() {
profile::reset_native();
}
let (outcome, measurement) =
measure(|| safe_trial(&config, &topic, prepared_group.as_deref()));
let broker_log_bytes = broker_log_bytes(&topic);
rows.push(ResultRow::new(
"warmup",
trial as isize,
&config,
measurement,
outcome,
broker_log_bytes,
));
}
let mut measured = Vec::new();
for trial in 0..config.iterations {
let topic = topic_name(&config.scenario, &format!("measurement-{trial}"));
create_topic(&config.bootstrap, &topic, config.partitions);
let prepared_group = prepare_trial(&config, &topic, "measurement", trial);
if profile::enabled() {
profile::reset();
}
if profile::native_enabled() {
profile::reset_native();
}
let (outcome, measurement) =
measure(|| safe_trial(&config, &topic, prepared_group.as_deref()));
let broker_log_bytes = broker_log_bytes(&topic);
let row = ResultRow::new(
"measurement",
trial as isize,
&config,
measurement,
outcome,
broker_log_bytes,
);
measured.push(row.clone());
rows.push(row);
}
let walls = measured
.iter()
.map(|row| row.measurement.wall_ns)
.collect::<Vec<_>>();
let mut summary = measured
.last()
.cloned()
.expect("at least one measurement iteration");
summary.phase = "summary".to_owned();
summary.trial = -1;
summary.wall_p50_ns = Some(percentile(&walls, 0.50));
summary.wall_p99_ns = Some(percentile(&walls, 0.99));
if measured.iter().any(|row| row.outcome.status != "ok") {
summary.outcome.status = "pending".to_owned();
summary.outcome.reason = measured
.iter()
.filter(|row| row.outcome.status != "ok")
.map(|row| row.outcome.reason.as_str())
.collect::<BTreeSet<_>>()
.into_iter()
.collect::<Vec<_>>()
.join("; ");
}
let profile_records = summary
.outcome
.consumed
.max(summary.outcome.produced)
.max(1);
let profile_measurement = summary.measurement.clone();
rows.push(summary);
emit_tsv(&config, &rows);
if profile::enabled() {
emit_profile_summary(profile_records, &profile_measurement);
}
if profile::native_enabled() {
emit_native_profile_summary(profile_records, &profile_measurement);
}
}
fn validate_scenario(scenario: &str) {
match scenario {
"produce_1M_256B"
| "consume_committable_1M"
| "e2e_latency_1M"
| "rebalance_disruption_1M" => {}
other => panic!("unknown scenario '{other}'"),
}
}
fn prepare_trial(config: &BenchConfig, topic: &str, phase: &str, trial: usize) -> Option<String> {
if config.scenario != "consume_committable_1M" {
return None;
}
let group = format!(
"{}-{}-{phase}-{trial}-{}",
config.client_id_prefix,
config.scenario,
now_ns()
);
initialize_group_offsets(config, topic, &group);
Some(group)
}
fn safe_trial(config: &BenchConfig, topic: &str, prepared_group: Option<&str>) -> TrialOutcome {
match std::panic::catch_unwind(|| trial(config, topic, prepared_group)) {
Ok(outcome) => outcome,
Err(_) => TrialOutcome::pending(
0,
0,
0,
0,
0,
end_offset_sum(&config.bootstrap, topic).unwrap_or(0),
"trial panicked",
),
}
}
fn trial(config: &BenchConfig, topic: &str, prepared_group: Option<&str>) -> TrialOutcome {
match config.scenario.as_str() {
"produce_1M_256B" => produce_trial(config, topic, false),
"consume_committable_1M" => {
split_measure(config, "seed", || seed_records(config, topic, false));
split_measure(config, "consume", || {
consume_trial(config, topic, false, true, prepared_group)
})
}
"e2e_latency_1M" => e2e_latency_trial(config, topic),
"rebalance_disruption_1M" => {
split_measure(config, "seed", || seed_records(config, topic, false));
split_measure(config, "rebalance", || rebalance_trial(config, topic))
}
_ => unreachable!(),
}
}
fn split_measure<T, F>(config: &BenchConfig, label: &str, body: F) -> T
where
F: FnOnce() -> T,
{
if !env_flag("MQ_BENCH_SPLITS") {
return body();
}
let before_cpu = process_cpu_ns();
let start = Instant::now();
let result = body();
let wall_ns = start.elapsed().as_nanos().max(1) as u64;
let cpu_ns = process_cpu_ns().saturating_sub(before_cpu);
eprintln!(
"MQ_BENCH_SPLIT\tclient={}\tscenario={}\tlabel={}\twall_ms={:.3}\tcpu_ms={:.3}",
config.client_name(),
config.scenario,
label,
wall_ns as f64 / 1_000_000.0,
cpu_ns as f64 / 1_000_000.0
);
result
}
fn produce_trial(config: &BenchConfig, topic: &str, latency_payload: bool) -> TrialOutcome {
produce_records(config, topic, latency_payload);
let end = end_offset_sum(&config.bootstrap, topic).unwrap_or(0);
if end == config.records as u64 {
TrialOutcome::ok(config.records as u64, 0, end, 0, end, end)
} else {
TrialOutcome::pending(
config.records as u64,
0,
end,
0,
end,
end,
&format!(
"topic end offsets {end} did not match produced {}",
config.records
),
)
}
}
fn produce_records(config: &BenchConfig, topic: &str, latency_payload: bool) {
let topic = topic.to_owned();
let payload_bytes = config.payload_bytes;
let records = config.records;
let source = Source::unfold(0_usize, move |index| {
if index >= records {
None
} else {
let sent_ns = if latency_payload { now_ns() as i64 } else { 0 };
let record = ProducerRecord::new(
topic.clone(),
Some(Bytes::from(payload(index, payload_bytes, sent_ns))),
)
.with_key(Some(Bytes::from((index as i64).to_be_bytes().to_vec())));
Some((index + 1, record))
}
});
let control = source
.run_with(KafkaSink::plain(producer_settings(config)))
.expect("producer materializes");
control.drain_and_shutdown().expect("producer drains");
}
fn seed_records(config: &BenchConfig, topic: &str, latency_payload: bool) {
let producer: BaseProducer = kafka_client_config(&config.bootstrap)
.set(
"client.id",
format!("{}-seed-{}", config.client_id_prefix, now_ns()),
)
.set("enable.idempotence", "false")
.set("acks", "1")
.set("linger.ms", "5")
.set("batch.num.messages", "10000")
.set("batch.size", "131072")
.set("queue.buffering.max.messages", "1000000")
.set("queue.buffering.max.kbytes", "1048576")
.set("max.in.flight.requests.per.connection", "1000000")
.set("compression.type", &config.compression)
.create()
.expect("seed producer");
let mut payload = vec![0_u8; config.payload_bytes.max(HEADER_BYTES)];
payload[16..24].copy_from_slice(&(config.payload_bytes as i64).to_be_bytes());
for index in 0..config.records {
let sent_ns = if latency_payload { now_ns() as i64 } else { 0 };
payload[0..8].copy_from_slice(&(index as i64).to_be_bytes());
payload[8..16].copy_from_slice(&sent_ns.to_be_bytes());
let partition = (index % config.partitions as usize) as i32;
loop {
match producer.send(
BaseRecord::<(), [u8], ()>::to(topic)
.payload(payload.as_slice())
.partition(partition),
) {
Ok(()) => break,
Err((
rdkafka::error::KafkaError::MessageProduction(RDKafkaErrorCode::QueueFull),
_,
)) => producer.poll(Duration::from_millis(1)),
Err((error, _)) => panic!("seed producer enqueue failed: {error}"),
}
}
if index % 1024 == 0 {
producer.poll(Duration::ZERO);
}
}
producer
.flush(Timeout::After(Duration::from_secs(30)))
.expect("seed producer final flush");
}
fn consume_trial(
config: &BenchConfig,
topic: &str,
capture_latency: bool,
verify_commit: bool,
prepared_group: Option<&str>,
) -> TrialOutcome {
let group = prepared_group.map_or_else(
|| {
format!(
"{}-{}-{}",
config.client_id_prefix,
config.scenario,
now_ns()
)
},
str::to_owned,
);
let counter = Arc::new(RecordCounter::new(config.records, capture_latency));
let (control, completion): (KafkaControl, StreamCompletion<datum::NotUsed>) =
split_measure(config, "consume_materialize", || {
KafkaSource::committable_payload_batches(
consumer_settings(config, &group, "consume"),
Subscription::topics([topic.to_owned()]),
)
.to_mat(
consumer_batch_sink(Arc::clone(&counter), config.consumer_backend),
Keep::both,
)
.run()
.expect("consumer materializes")
});
let assigned = split_measure(config, "consume_assignment", || {
wait_until(Duration::from_secs(30), || {
control.metrics().snapshot().assigned_partitions >= config.partitions as u64
})
});
if !assigned {
control.shutdown_now();
completion.wait().expect("consumer completes");
return TrialOutcome::pending(
config.records as u64,
counter.consumed(),
counter.unique_count(),
counter.duplicates(),
0,
end_offset_sum(&config.bootstrap, topic).unwrap_or(0),
"consumer assignment did not reach all partitions before consume deadline",
);
}
let reached = split_measure(config, "consume_wait", || {
wait_until(Duration::from_secs(120), || {
counter.unique_count() >= config.records as u64
})
});
if reached {
split_measure(config, "consume_drain_shutdown", || {
control
.drain_and_shutdown(Duration::from_secs(30))
.expect("consumer drains");
});
} else {
control.shutdown_now();
}
split_measure(config, "consume_completion", || {
completion.wait().expect("consumer completes");
});
let offsets = split_measure(config, "consume_offsets", || {
if verify_commit {
wait_for_offsets(&config.bootstrap, topic, &group, config.partitions)
} else {
OffsetSummary {
end_offsets: end_offset_sum(&config.bootstrap, topic).unwrap_or(0),
committed_offsets: 0,
}
}
});
let mut outcome = outcome_from_counter(config, &counter, offsets, None);
if !reached {
outcome.status = "pending".to_owned();
outcome.reason = "consumer did not observe all records before timeout".to_owned();
}
outcome
}
fn initialize_group_offsets(config: &BenchConfig, topic: &str, group: &str) {
let consumer = group_consumer(&config.bootstrap, group).expect("benchmark group consumer");
let mut offsets = TopicPartitionList::with_capacity(config.partitions as usize);
for partition in 0..config.partitions {
offsets
.add_partition_offset(topic, partition, Offset::Offset(0))
.expect("benchmark group partition offset");
}
consumer
.commit(&offsets, CommitMode::Sync)
.expect("initialize benchmark group offsets");
}
fn e2e_latency_trial(config: &BenchConfig, topic: &str) -> TrialOutcome {
let group = format!("{}-latency-{}", config.client_id_prefix, now_ns());
let counter = Arc::new(RecordCounter::new(config.records, true));
let (control, completion): (KafkaControl, StreamCompletion<datum::NotUsed>) =
KafkaSource::committable_payload_batches(
consumer_settings(config, &group, "latency"),
Subscription::topics([topic.to_owned()]),
)
.to_mat(
consumer_batch_sink(Arc::clone(&counter), config.consumer_backend),
Keep::both,
)
.run()
.expect("latency consumer materializes");
let assigned = wait_until(Duration::from_secs(30), || {
control.metrics().snapshot().assigned_partitions >= config.partitions as u64
});
if !assigned {
control.shutdown_now();
return TrialOutcome::pending(
0,
counter.consumed(),
counter.unique_count(),
counter.duplicates(),
0,
end_offset_sum(&config.bootstrap, topic).unwrap_or(0),
"consumer assignment did not reach all partitions before producing",
);
}
let producer = produce_trial(config, topic, true);
if producer.status != "ok" {
control.shutdown_now();
let reason = producer.reason.clone();
return producer.with_reason(format!(
"latency producer did not complete cleanly: {}",
reason
));
}
let reached = wait_until(Duration::from_secs(120), || {
counter.unique_count() >= config.records as u64
});
if reached {
control
.drain_and_shutdown(Duration::from_secs(30))
.expect("latency consumer drains");
} else {
control.shutdown_now();
}
completion.wait().expect("latency consumer completes");
let offsets = wait_for_offsets(&config.bootstrap, topic, &group, config.partitions);
let mut outcome = outcome_from_counter(config, &counter, offsets, None);
if !reached {
outcome.status = "pending".to_owned();
outcome.reason = "latency consumer did not observe all records before timeout".to_owned();
}
outcome
}
fn rebalance_trial(config: &BenchConfig, topic: &str) -> TrialOutcome {
let group = format!("{}-rebalance-{}", config.client_id_prefix, now_ns());
let counter = Arc::new(RecordCounter::new(config.records, false));
let disruption_marked = Arc::new(AtomicBool::new(false));
let mut controls = Vec::new();
let mut completions = Vec::new();
for member in 0..4 {
let counter_for_sink = Arc::clone(&counter);
let disruption_for_sink = Arc::clone(&disruption_marked);
let backend = config.consumer_backend;
let (control, completion) = KafkaSource::committable_payload_batches(
consumer_settings(config, &group, &format!("rebalance-{member}")),
Subscription::topics([topic.to_owned()]),
)
.to_mat(
Sink::foreach_result(move |batch: KafkaPayloadBatch| {
measure_datum_consume(backend, || {
counter_for_sink.observe_payload_batch(&batch);
if disruption_for_sink.load(Ordering::SeqCst) {
counter_for_sink.mark_disruption_seen();
}
});
batch.commit().map_err(StreamError::from)?;
Ok(())
}),
Keep::both,
)
.run()
.expect("rebalance consumer materializes");
controls.push(control);
completions.push(completion);
}
let reached_mid = wait_until(Duration::from_secs(30), || {
counter.consumed() >= (config.records / 4).max(1) as u64
});
counter.mark_disruption();
disruption_marked.store(true, Ordering::SeqCst);
if let Some(control) = controls.first() {
control.shutdown_now();
}
let reached_end = wait_until(Duration::from_secs(120), || {
counter.unique_count() >= config.records as u64
});
for control in controls.iter().skip(1) {
if reached_end {
let _ = control.drain_and_shutdown(Duration::from_secs(30));
} else {
control.shutdown_now();
}
}
for completion in completions {
let _ = completion.wait();
}
let offsets = wait_for_offsets(&config.bootstrap, topic, &group, config.partitions);
let mut outcome = outcome_from_counter(config, &counter, offsets, counter.rebalance_pause_ns());
if !reached_mid {
outcome.status = "pending".to_owned();
outcome.reason = "did not reach mid-stream threshold before member kill".to_owned();
} else if !reached_end && counter.unique_count() < config.records as u64 {
outcome.status = "pending".to_owned();
outcome.reason = "remaining consumers did not consume all records after member kill; lever: increase consumer parallelism or lower commit latency".to_owned();
}
outcome
}
fn consumer_batch_sink(
counter: Arc<RecordCounter>,
backend: KafkaConsumerBackend,
) -> Sink<KafkaPayloadBatch, StreamCompletion<datum::NotUsed>> {
Sink::foreach_result(move |batch: KafkaPayloadBatch| {
measure_datum_consume(backend, || {
counter.observe_payload_batch(&batch);
});
batch.commit().map_err(StreamError::from)?;
Ok(())
})
}
fn measure_datum_consume<T, F>(backend: KafkaConsumerBackend, body: F) -> T
where
F: FnOnce() -> T,
{
match backend {
KafkaConsumerBackend::Rdkafka => profile::measure(ProfileBucket::DatumConsume, body),
KafkaConsumerBackend::Native => profile::measure_native_datum_consume(body),
}
}
fn producer_settings(config: &BenchConfig) -> KafkaProducerSettings {
let mut settings = KafkaProducerSettings::new(config.bootstrap.clone())
.with_producer_backend(config.producer_backend)
.with(
"client.id",
format!("{}-producer-{}", config.client_id_prefix, now_ns()),
)
.with("linger.ms", "5")
.with("batch.num.messages", "10000")
.with("batch.size", "131072")
.with("queue.buffering.max.messages", "1000000")
.with("queue.buffering.max.kbytes", "1048576")
.with("compression.type", config.compression.clone())
.with_in_flight_limit(524_288);
settings = apply_security_settings(settings);
if config.producer_backend == KafkaProducerBackend::Native {
settings =
settings.with_max_in_flight_requests_per_connection(config.producer_max_in_flight);
}
if matches!(
config.scenario.as_str(),
"consume_committable_1M" | "rebalance_disruption_1M"
) {
settings = settings
.with("enable.idempotence", "false")
.with("acks", "1")
.with("max.in.flight.requests.per.connection", "1000000");
} else if config.scenario == "e2e_latency_1M" {
settings = settings
.with("linger.ms", "0")
.with("queue.buffering.max.messages", "8192")
.with("queue.buffering.max.kbytes", "65536")
.with_in_flight_limit(1024);
}
if config.producer_backend == KafkaProducerBackend::Native
&& let Some(enabled) = config.producer_enable_idempotence
{
settings = settings.with("enable.idempotence", enabled.to_string());
if enabled {
settings = settings.with("acks", "all");
}
}
settings
}
fn consumer_settings(config: &BenchConfig, group: &str, suffix: &str) -> KafkaConsumerSettings {
let mut settings = KafkaConsumerSettings::new(config.bootstrap.clone(), group.to_owned())
.with(
"client.id",
format!("{}-{suffix}-{}", config.client_id_prefix, now_ns()),
)
.with("auto.offset.reset", "earliest")
.with("fetch.min.bytes", "1")
.with("fetch.wait.max.ms", "500")
.with("fetch.max.bytes", "52428800")
.with("fetch.message.max.bytes", "1048576")
.with("queued.min.messages", "1000000")
.with("queued.max.messages.kbytes", "1048576")
.with("fetch.queue.backoff.ms", "1")
.with("session.timeout.ms", "45000")
.with("heartbeat.interval.ms", "3000")
.with("partition.assignment.strategy", "cooperative-sticky")
.with_backpressure(
config.commit_batch,
config.commit_batch.saturating_mul(2).max(2),
)
.with_poll_batch_size(10_000)
.with_payload_timestamps(false)
.with_commit_batch_size(config.commit_batch)
.with_commit_interval(Duration::from_millis(50))
.with_consumer_backend(config.consumer_backend);
settings = apply_consumer_security_settings(settings);
if config.scenario == "e2e_latency_1M" {
settings = settings
.with_low_latency(true)
.with_poll_batch_size(4096)
.with("fetch.wait.max.ms", "1");
} else if config.scenario == "consume_committable_1M" {
settings = settings.with(
"fetch.min.bytes",
env::var("MQ_FETCH_MIN_BYTES").unwrap_or_else(|_| "1048576".to_owned()),
);
}
settings.poll_timeout = Duration::from_millis(1);
settings.commit_sync = false;
settings
}
#[derive(Debug)]
struct RecordCounter {
expected: usize,
seen: SeenSet,
unique: AtomicU64,
consumed: AtomicU64,
duplicates: AtomicU64,
capture_latency: bool,
latencies: Mutex<Vec<u64>>,
disruption_ns: AtomicU64,
rebalance_pause_ns: AtomicU64,
}
#[derive(Debug)]
enum SeenSet {
Locked(Mutex<Vec<u8>>),
Atomic(Vec<AtomicU8>),
}
impl RecordCounter {
fn new(expected: usize, capture_latency: bool) -> Self {
let seen = if capture_latency {
SeenSet::Atomic((0..expected).map(|_| AtomicU8::new(0)).collect())
} else {
SeenSet::Locked(Mutex::new(vec![0; expected]))
};
Self {
expected,
seen,
unique: AtomicU64::new(0),
consumed: AtomicU64::new(0),
duplicates: AtomicU64::new(0),
capture_latency,
latencies: Mutex::new(if capture_latency {
Vec::with_capacity(expected)
} else {
Vec::new()
}),
disruption_ns: AtomicU64::new(0),
rebalance_pause_ns: AtomicU64::new(0),
}
}
fn observe_payload_batch(&self, batch: &KafkaPayloadBatch) {
match &self.seen {
SeenSet::Locked(seen) => self.observe_payload_batch_locked(batch, seen),
SeenSet::Atomic(seen) => self.observe_payload_batch_atomic(batch, seen),
}
}
fn observe_payload_batch_locked(&self, batch: &KafkaPayloadBatch, seen: &Mutex<Vec<u8>>) {
let mut seen = seen.lock().expect("seen lock");
let mut latencies = if self.capture_latency {
Some(self.latencies.lock().expect("latency lock"))
} else {
None
};
let mut consumed = 0_u64;
let mut unique = 0_u64;
let mut duplicates = 0_u64;
for record in batch.records() {
consumed += 1;
let payload = batch.payload(record);
let seq = payload_seq(payload);
if seq < self.expected {
if seen[seq] != 0 {
duplicates += 1;
} else {
seen[seq] = 1;
unique += 1;
}
}
if let Some(latencies) = latencies.as_mut() {
let sent = payload_sent_ns(payload);
if sent > 0 {
latencies.push(now_ns().saturating_sub(sent));
}
}
}
self.consumed.fetch_add(consumed, Ordering::Relaxed);
if unique > 0 {
self.unique.fetch_add(unique, Ordering::Relaxed);
}
if duplicates > 0 {
self.duplicates.fetch_add(duplicates, Ordering::Relaxed);
}
}
fn observe_payload_batch_atomic(&self, batch: &KafkaPayloadBatch, seen: &[AtomicU8]) {
let mut latencies = if self.capture_latency {
Some(self.latencies.lock().expect("latency lock"))
} else {
None
};
let mut consumed = 0_u64;
let mut unique = 0_u64;
let mut duplicates = 0_u64;
for record in batch.records() {
consumed += 1;
let payload = batch.payload(record);
let seq = payload_seq(payload);
if seq < self.expected {
if seen[seq].swap(1, Ordering::Relaxed) != 0 {
duplicates += 1;
} else {
unique += 1;
}
}
if let Some(latencies) = latencies.as_mut() {
let sent = payload_sent_ns(payload);
if sent > 0 {
latencies.push(now_ns().saturating_sub(sent));
}
}
}
self.consumed.fetch_add(consumed, Ordering::Relaxed);
if unique > 0 {
self.unique.fetch_add(unique, Ordering::Relaxed);
}
if duplicates > 0 {
self.duplicates.fetch_add(duplicates, Ordering::Relaxed);
}
}
fn mark_disruption(&self) {
let _ =
self.disruption_ns
.compare_exchange(0, now_ns(), Ordering::SeqCst, Ordering::SeqCst);
}
fn mark_disruption_seen(&self) {
let started = self.disruption_ns.load(Ordering::SeqCst);
if started > 0 {
let pause = now_ns().saturating_sub(started);
let _ = self.rebalance_pause_ns.compare_exchange(
0,
pause,
Ordering::SeqCst,
Ordering::SeqCst,
);
}
}
fn consumed(&self) -> u64 {
self.consumed.load(Ordering::Relaxed)
}
fn duplicates(&self) -> u64 {
self.duplicates.load(Ordering::Relaxed)
}
fn unique_count(&self) -> u64 {
self.unique.load(Ordering::Relaxed)
}
fn latency_p50_p99(&self) -> (Option<u64>, Option<u64>) {
let latencies = self.latencies.lock().expect("latency lock");
if latencies.is_empty() {
(None, None)
} else {
(
Some(percentile(&latencies, 0.50)),
Some(percentile(&latencies, 0.99)),
)
}
}
fn rebalance_pause_ns(&self) -> Option<u64> {
match self.rebalance_pause_ns.load(Ordering::Relaxed) {
0 => None,
value => Some(value),
}
}
}
fn outcome_from_counter(
config: &BenchConfig,
counter: &RecordCounter,
offsets: OffsetSummary,
rebalance_pause_ns: Option<u64>,
) -> TrialOutcome {
let (latency_p50_ns, latency_p99_ns) = counter.latency_p50_p99();
let ok = counter.unique_count() == config.records as u64
&& offsets.end_offsets == config.records as u64
&& offsets.commit_ok();
let reason = if ok {
String::new()
} else {
format!(
"unique={}, consumed={}, duplicates={}, end_offsets={}, committed_offsets={}; lever: batch commits and rebalance drain tuning",
counter.unique_count(),
counter.consumed(),
counter.duplicates(),
offsets.end_offsets,
offsets.committed_offsets
)
};
TrialOutcome {
produced: config.records as u64,
consumed: counter.consumed(),
unique_consumed: counter.unique_count(),
duplicates: counter.duplicates(),
committed_offsets: offsets.committed_offsets,
end_offsets: offsets.end_offsets,
latency_p50_ns,
latency_p99_ns,
rebalance_pause_ns,
status: if ok { "ok" } else { "pending" }.to_owned(),
reason,
}
}
#[derive(Debug, Clone)]
struct OffsetSummary {
end_offsets: u64,
committed_offsets: u64,
}
impl OffsetSummary {
fn commit_ok(&self) -> bool {
self.end_offsets == self.committed_offsets
}
}
fn wait_for_offsets(
bootstrap: &str,
topic: &str,
group: &str,
partition_count: i32,
) -> OffsetSummary {
let partitions = (0..partition_count).collect::<Vec<_>>();
let mut current = offsets(bootstrap, topic, group, &partitions);
wait_until(Duration::from_secs(30), || {
current = offsets(bootstrap, topic, group, &partitions);
current.commit_ok()
});
current
}
fn offsets(bootstrap: &str, topic: &str, group: &str, partitions: &[i32]) -> OffsetSummary {
let Some(consumer) = group_consumer(bootstrap, group) else {
return OffsetSummary {
end_offsets: 0,
committed_offsets: 0,
};
};
OffsetSummary {
end_offsets: end_offset_sum_for(&consumer, topic, partitions).unwrap_or(0),
committed_offsets: committed_offset_sum_for(&consumer, topic, partitions).unwrap_or(0),
}
}
fn create_topic(bootstrap: &str, topic: &str, partitions: i32) {
let admin: AdminClient<DefaultClientContext> = kafka_client_config(bootstrap)
.create()
.expect("admin client");
let new_topic =
NewTopic::new(topic, partitions, TopicReplication::Fixed(1)).set("retention.ms", "600000");
let results = block_on(admin.create_topics(&[new_topic], &AdminOptions::new()))
.expect("topic create response");
for result in results {
match result {
Ok(_) => {}
Err((_, RDKafkaErrorCode::TopicAlreadyExists)) => {}
Err((name, error)) => panic!("failed to create topic {name}: {error:?}"),
}
}
wait_until(Duration::from_secs(30), || {
topic_partitions(bootstrap, topic).len() >= partitions as usize
});
}
fn end_offset_sum(bootstrap: &str, topic: &str) -> Option<u64> {
let consumer = metadata_consumer(bootstrap, "end-offsets");
let partitions = topic_partitions_for(&consumer, topic);
end_offset_sum_for(&consumer, topic, &partitions)
}
fn end_offset_sum_for(consumer: &BaseConsumer, topic: &str, partitions: &[i32]) -> Option<u64> {
let mut sum = 0_u64;
for &partition in partitions {
let (_, high) = consumer
.fetch_watermarks(topic, partition, Timeout::After(Duration::from_secs(5)))
.ok()?;
sum = sum.saturating_add(high.max(0) as u64);
}
Some(sum)
}
fn committed_offset_sum_for(
consumer: &BaseConsumer,
topic: &str,
partitions: &[i32],
) -> Option<u64> {
let mut tpl = TopicPartitionList::new();
for &partition in partitions {
tpl.add_partition(topic, partition);
}
let committed = consumer
.committed_offsets(tpl, Timeout::After(Duration::from_secs(5)))
.ok()?;
Some(
committed
.elements()
.into_iter()
.filter_map(|element| match element.offset() {
Offset::Offset(offset) if offset > 0 => Some(offset as u64),
_ => None,
})
.sum(),
)
}
fn topic_partitions(bootstrap: &str, topic: &str) -> Vec<i32> {
let consumer = metadata_consumer(bootstrap, "metadata");
topic_partitions_for(&consumer, topic)
}
fn topic_partitions_for(consumer: &BaseConsumer, topic: &str) -> Vec<i32> {
consumer
.fetch_metadata(Some(topic), Timeout::After(Duration::from_secs(5)))
.ok()
.and_then(|metadata| {
metadata
.topics()
.iter()
.find(|metadata_topic| metadata_topic.name() == topic)
.map(|metadata_topic| {
metadata_topic
.partitions()
.iter()
.map(|partition| partition.id())
.collect::<Vec<_>>()
})
})
.unwrap_or_default()
}
fn group_consumer(bootstrap: &str, group: &str) -> Option<BaseConsumer> {
kafka_client_config(bootstrap)
.set("group.id", group)
.set("enable.auto.commit", "false")
.create()
.ok()
}
fn metadata_consumer(bootstrap: &str, suffix: &str) -> BaseConsumer {
kafka_client_config(bootstrap)
.set("group.id", format!("datum-mq-{suffix}-{}", now_ns()))
.set("enable.auto.commit", "false")
.create()
.expect("metadata consumer")
}
fn kafka_client_config(bootstrap: &str) -> ClientConfig {
let mut config = ClientConfig::new();
config.set("bootstrap.servers", bootstrap);
for (environment, property) in security_environment() {
if let Ok(value) = env::var(environment) {
config.set(property, value);
}
}
config
}
fn apply_security_settings(mut settings: KafkaProducerSettings) -> KafkaProducerSettings {
for (environment, property) in security_environment() {
if let Ok(value) = env::var(environment) {
settings = settings.with(property, value);
}
}
settings
}
fn apply_consumer_security_settings(mut settings: KafkaConsumerSettings) -> KafkaConsumerSettings {
for (environment, property) in security_environment() {
if let Ok(value) = env::var(environment) {
settings = settings.with(property, value);
}
}
settings
}
fn security_environment() -> [(&'static str, &'static str); 5] {
[
("MQ_SECURITY_PROTOCOL", "security.protocol"),
("MQ_SSL_CA_LOCATION", "ssl.ca.location"),
("MQ_SASL_MECHANISM", "sasl.mechanism"),
("MQ_SASL_USERNAME", "sasl.username"),
("MQ_SASL_PASSWORD", "sasl.password"),
]
}
fn payload(seq: usize, payload_bytes: usize, sent_ns: i64) -> Vec<u8> {
let mut bytes = vec![0_u8; payload_bytes.max(HEADER_BYTES)];
bytes[0..8].copy_from_slice(&(seq as i64).to_be_bytes());
bytes[8..16].copy_from_slice(&sent_ns.to_be_bytes());
bytes[16..24].copy_from_slice(&(payload_bytes as i64).to_be_bytes());
for (index, byte) in bytes.iter_mut().enumerate().skip(HEADER_BYTES) {
*byte = seq.wrapping_add(index) as u8;
}
bytes
}
fn payload_seq(payload: &[u8]) -> usize {
if payload.len() < 8 {
usize::MAX
} else {
i64::from_be_bytes(payload[0..8].try_into().expect("slice len")) as usize
}
}
fn payload_sent_ns(payload: &[u8]) -> u64 {
if payload.len() < 16 {
0
} else {
i64::from_be_bytes(payload[8..16].try_into().expect("slice len")).max(0) as u64
}
}
#[derive(Debug, Clone)]
struct ProcStatus {
peak_rss_kb: u64,
current_rss_kb: u64,
threads: u64,
}
#[derive(Debug, Clone)]
struct Measurement {
wall_ns: u64,
cpu_ns: u64,
status: ProcStatus,
}
fn measure<F>(body: F) -> (TrialOutcome, Measurement)
where
F: FnOnce() -> TrialOutcome,
{
let before_cpu = process_cpu_ns();
let start = Instant::now();
let outcome = body();
let wall_ns = start.elapsed().as_nanos().max(1) as u64;
let cpu_ns = process_cpu_ns().saturating_sub(before_cpu);
(
outcome,
Measurement {
wall_ns,
cpu_ns,
status: process_status(),
},
)
}
fn process_cpu_ns() -> u64 {
let Ok(stat) = fs::read_to_string("/proc/self/stat") else {
return 0;
};
let Some(close) = stat.rfind(')') else {
return 0;
};
let fields = stat[close + 1..].split_whitespace().collect::<Vec<_>>();
if fields.len() <= 12 {
return 0;
}
let utime = fields[11].parse::<u64>().unwrap_or(0);
let stime = fields[12].parse::<u64>().unwrap_or(0);
((utime + stime) as f64 * 1_000_000_000.0 / clock_ticks_per_second()) as u64
}
fn clock_ticks_per_second() -> f64 {
Command::new("getconf")
.arg("CLK_TCK")
.output()
.ok()
.and_then(|output| String::from_utf8(output.stdout).ok())
.and_then(|value| value.trim().parse::<f64>().ok())
.unwrap_or(100.0)
}
fn process_status() -> ProcStatus {
let mut status = ProcStatus {
peak_rss_kb: 0,
current_rss_kb: 0,
threads: 0,
};
if let Ok(text) = fs::read_to_string("/proc/self/status") {
for line in text.lines() {
if line.starts_with("VmHWM:") {
status.peak_rss_kb = first_number(line);
} else if line.starts_with("VmRSS:") {
status.current_rss_kb = first_number(line);
} else if line.starts_with("Threads:") {
status.threads = first_number(line);
}
}
}
status
}
fn first_number(line: &str) -> u64 {
line.split_whitespace()
.find_map(|part| part.parse::<u64>().ok())
.unwrap_or(0)
}
#[derive(Debug, Clone)]
struct TrialOutcome {
produced: u64,
consumed: u64,
unique_consumed: u64,
duplicates: u64,
committed_offsets: u64,
end_offsets: u64,
latency_p50_ns: Option<u64>,
latency_p99_ns: Option<u64>,
rebalance_pause_ns: Option<u64>,
status: String,
reason: String,
}
impl TrialOutcome {
fn ok(
produced: u64,
consumed: u64,
unique_consumed: u64,
duplicates: u64,
committed_offsets: u64,
end_offsets: u64,
) -> Self {
Self {
produced,
consumed,
unique_consumed,
duplicates,
committed_offsets,
end_offsets,
latency_p50_ns: None,
latency_p99_ns: None,
rebalance_pause_ns: None,
status: "ok".to_owned(),
reason: String::new(),
}
}
fn pending(
produced: u64,
consumed: u64,
unique_consumed: u64,
duplicates: u64,
committed_offsets: u64,
end_offsets: u64,
reason: &str,
) -> Self {
Self {
status: "pending".to_owned(),
reason: reason.to_owned(),
..Self::ok(
produced,
consumed,
unique_consumed,
duplicates,
committed_offsets,
end_offsets,
)
}
}
fn commit_ok(&self) -> bool {
self.committed_offsets == self.end_offsets
}
fn no_loss(&self) -> bool {
self.unique_consumed >= self.end_offsets
}
fn with_reason(mut self, reason: String) -> Self {
self.status = "pending".to_owned();
self.reason = reason;
self
}
}
#[derive(Debug, Clone)]
struct ResultRow {
client: String,
scenario: String,
phase: String,
trial: isize,
records: usize,
payload_bytes: usize,
partitions: i32,
compression: String,
broker_log_bytes: Option<u64>,
measurement: Measurement,
outcome: TrialOutcome,
wall_p50_ns: Option<u64>,
wall_p99_ns: Option<u64>,
}
impl ResultRow {
fn new(
phase: &str,
trial: isize,
config: &BenchConfig,
measurement: Measurement,
outcome: TrialOutcome,
broker_log_bytes: Option<u64>,
) -> Self {
Self {
client: config.client_name().to_owned(),
scenario: config.scenario.clone(),
phase: phase.to_owned(),
trial,
records: config.records,
payload_bytes: config.payload_bytes,
partitions: config.partitions,
compression: config.compression.clone(),
broker_log_bytes,
measurement,
outcome,
wall_p50_ns: None,
wall_p99_ns: None,
}
}
fn fields(&self) -> Vec<String> {
let records = self.outcome.produced.max(self.outcome.consumed).max(1);
let records_per_s = records as f64 / (self.measurement.wall_ns as f64 / 1e9);
let mb_per_s = records_per_s * self.payload_bytes as f64 / 1_000_000.0;
let cpu_cores = self.measurement.cpu_ns as f64 / self.measurement.wall_ns as f64;
vec![
self.client.clone(),
self.scenario.clone(),
self.phase.clone(),
self.trial.to_string(),
self.records.to_string(),
self.payload_bytes.to_string(),
self.partitions.to_string(),
self.compression.clone(),
self.broker_log_bytes
.map(|bytes| bytes.to_string())
.unwrap_or_default(),
self.broker_log_bytes
.map(|bytes| {
let raw = self.records.saturating_mul(self.payload_bytes).max(1) as f64;
format!("{:.4}", bytes as f64 / raw)
})
.unwrap_or_default(),
format_ms(self.measurement.wall_ns),
self.wall_p50_ns.map(format_ms).unwrap_or_default(),
self.wall_p99_ns.map(format_ms).unwrap_or_default(),
format!("{records_per_s:.2}"),
format!("{mb_per_s:.2}"),
format_ms(self.measurement.cpu_ns),
format!("{cpu_cores:.3}"),
"pending".to_owned(),
"pending".to_owned(),
self.measurement.status.peak_rss_kb.to_string(),
self.measurement.status.current_rss_kb.to_string(),
self.measurement.status.threads.to_string(),
self.outcome.produced.to_string(),
self.outcome.consumed.to_string(),
self.outcome.unique_consumed.to_string(),
self.outcome.duplicates.to_string(),
self.outcome.committed_offsets.to_string(),
self.outcome.end_offsets.to_string(),
self.outcome.commit_ok().to_string(),
self.outcome.no_loss().to_string(),
self.outcome
.latency_p50_ns
.map(format_ms)
.unwrap_or_default(),
self.outcome
.latency_p99_ns
.map(format_ms)
.unwrap_or_default(),
self.outcome
.rebalance_pause_ns
.map(format_ms)
.unwrap_or_default(),
self.outcome.status.clone(),
self.outcome.reason.replace(['\t', '\n'], " "),
]
}
}
fn emit_tsv(config: &BenchConfig, rows: &[ResultRow]) {
let mut text = String::new();
text.push_str(&header().join("\t"));
text.push('\n');
for row in rows {
text.push_str(&row.fields().join("\t"));
text.push('\n');
}
print!("{text}");
if let Some(path) = &config.out {
if let Some(parent) = path.parent() {
fs::create_dir_all(parent).expect("create benchmark output dir");
}
fs::write(path, text).expect("write benchmark output");
}
}
fn broker_log_bytes(topic: &str) -> Option<u64> {
let data_dir = env::var_os("MQ_BROKER_DATA_DIR")
.map(PathBuf::from)
.unwrap_or_else(|| {
PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../../baselines/mq/.runtime/kraft-data")
});
let prefix = format!("{topic}-");
let mut total = 0_u64;
let mut found = false;
for entry in fs::read_dir(data_dir).ok()? {
let entry = entry.ok()?;
if !entry.file_type().ok()?.is_dir()
|| !entry.file_name().to_string_lossy().starts_with(&prefix)
{
continue;
}
for file in fs::read_dir(entry.path()).ok()? {
let file = file.ok()?;
if file.path().extension().and_then(|value| value.to_str()) == Some("log") {
total = total.saturating_add(file.metadata().ok()?.len());
found = true;
}
}
}
found.then_some(total)
}
fn emit_profile_summary(records: u64, measurement: &Measurement) {
let snapshot = profile::snapshot();
eprintln!();
eprintln!(
"DATUM_MQ_PROFILE consume-path breakdown for {} records (process wall {}, process CPU {}, source handoff channel: std::sync::mpsc::sync_channel(1))",
records,
format_ms(measurement.wall_ns),
format_ms(measurement.cpu_ns),
);
eprintln!(
"| Path component | Calls | Wall ms | Thread CPU ms | Wall ns/record | CPU ns/record | Note |"
);
eprintln!("| --- | ---: | ---: | ---: | ---: | ---: | --- |");
for (name, bucket, note) in profile_rows(snapshot) {
eprintln!(
"| {name} | {} | {} | {} | {:.1} | {:.1} | {note} |",
bucket.calls,
format_ms(bucket.wall_ns),
format_ms(bucket.cpu_ns),
bucket.wall_ns as f64 / records as f64,
bucket.cpu_ns as f64 / records as f64,
);
}
}
fn emit_native_profile_summary(records: u64, measurement: &Measurement) {
eprintln!();
eprintln!(
"DATUM_KAFKA_NATIVE_PROFILE consume-path breakdown for {} records (process wall {}, process CPU {}, source handoff channel: std::sync::mpsc::sync_channel(1))",
records,
format_ms(measurement.wall_ns),
format_ms(measurement.cpu_ns),
);
eprintln!(
"| Path component | Calls | Wall ms | Thread CPU ms | Wall ns/record | CPU ns/record | Note |"
);
eprintln!("| --- | ---: | ---: | ---: | ---: | ---: | --- |");
for (name, bucket, note) in profile::native_profile_rows() {
eprintln!(
"| {name} | {} | {} | {} | {:.1} | {:.1} | {note} |",
bucket.calls,
format_ms(bucket.wall_ns),
format_ms(bucket.cpu_ns),
bucket.wall_ns as f64 / records as f64,
bucket.cpu_ns as f64 / records as f64,
);
}
}
fn profile_rows(
snapshot: ProfileSnapshot,
) -> [(&'static str, ProfileBucketSnapshot, &'static str); 6] {
[
(
"rdkafka poll loop / fetch",
snapshot.poll_fetch,
"BaseConsumer::poll on datum-mq-consumer-poll",
),
(
"BorrowedMessage -> ConsumerRecord",
snapshot.record_conversion,
"payload batch copy or full owned conversion",
),
(
"poller -> Source channel handoff",
snapshot.channel_handoff,
"std sync_channel send/try_send",
),
(
"Datum Source emit/envelope",
snapshot.source_emit,
"resource read plus offset handle creation",
),
(
"Datum sink consumption",
snapshot.datum_consume,
"benchmark counter over payload batch",
),
(
"commit bookkeeping",
snapshot.commit_bookkeeping,
"offset watermark tracking and commit trigger",
),
]
}
fn header() -> Vec<&'static str> {
vec![
"client",
"scenario",
"phase",
"trial",
"records",
"payload_bytes",
"partitions",
"compression",
"broker_log_bytes",
"broker_log_ratio",
"wall_ms",
"wall_p50_ms",
"wall_p99_ms",
"records_per_s",
"mb_per_s",
"cpu_ms",
"cpu_cores",
"alloc_bytes",
"alloc_bytes_per_record",
"rss_peak_kib",
"rss_current_kib",
"threads",
"produced",
"consumed",
"unique_consumed",
"duplicates",
"committed_offsets",
"end_offsets",
"commit_ok",
"no_loss",
"latency_p50_ms",
"latency_p99_ms",
"rebalance_pause_ms",
"status",
"reason",
]
}
fn percentile(values: &[u64], quantile: f64) -> u64 {
if values.is_empty() {
return 0;
}
let mut values = values.to_vec();
values.sort_unstable();
let index = ((values.len() as f64 * quantile).ceil() as usize)
.saturating_sub(1)
.min(values.len() - 1);
values[index]
}
fn wait_until<F>(timeout: Duration, mut predicate: F) -> bool
where
F: FnMut() -> bool,
{
let deadline = Instant::now() + timeout;
while Instant::now() < deadline {
if predicate() {
return true;
}
thread::sleep(Duration::from_millis(100));
}
predicate()
}
fn topic_name(scenario: &str, suffix: &str) -> String {
format!(
"datum_mq_datum_{scenario}_{}_{}_{}",
millis(),
suffix,
now_ns()
)
.chars()
.map(|ch| {
if ch.is_ascii_alphanumeric() || ch == '_' || ch == '-' || ch == '.' {
ch
} else {
'_'
}
})
.collect()
}
fn format_ms(ns: u64) -> String {
format!("{:.3}", ns as f64 / 1_000_000.0)
}
fn now_ns() -> u64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_nanos() as u64
}
fn millis() -> u128 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_millis()
}