You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by gk...@apache.org on 2025/08/07 07:45:10 UTC
(iggy) branch master updated: feat(bench): enable high-level API for producer and producing consumer (#1919)
This is an automated email from the ASF dual-hosted git repository.
gkoszyk pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 1df72545 feat(bench): enable high-level API for producer and producing consumer (#1919)
1df72545 is described below
commit 1df72545d6303d26fde8f34474f5435a48a2ad23
Author: WangYang <50...@users.noreply.github.com>
AuthorDate: Thu Aug 7 09:45:06 2025 +0200
feat(bench): enable high-level API for producer and producing consumer (#1919)
Add HighLevelBackend implementations for producer and producing consumer
benchmarks.
This enables the `--high-level-api` flag to be used across all benchmark
types, completing the support started with the high-level consumer
backend.
Closes #1856
---------
Co-authored-by: Yang Wong <yang wang>
---
core/bench/src/actors/consumer/backend.rs | 84 ---
.../src/actors/consumer/benchmark_consumer.rs | 171 +++----
.../high_level.rs} | 180 +++----
.../mod.rs => consumer/client/interface.rs} | 21 +-
core/bench/src/actors/consumer/client/low_level.rs | 162 ++++++
.../actors/{producer => consumer/client}/mod.rs | 6 +-
.../bench/src/actors/consumer/low_level_backend.rs | 201 --------
core/bench/src/actors/consumer/mod.rs | 6 +-
.../actors/consumer/typed_benchmark_consumer.rs | 100 ++++
core/bench/src/actors/mod.rs | 20 +
.../src/actors/producer/benchmark_producer.rs | 186 ++++---
.../bench/src/actors/producer/client/high_level.rs | 103 ++++
.../mod.rs => producer/client/interface.rs} | 26 +-
core/bench/src/actors/producer/client/low_level.rs | 113 +++++
core/bench/src/actors/producer/{ => client}/mod.rs | 6 +-
core/bench/src/actors/producer/mod.rs | 3 +-
.../actors/producer/typed_benchmark_producer.rs | 99 ++++
.../benchmark_producing_consumer.rs | 563 +++++++--------------
core/bench/src/actors/producing_consumer/mod.rs | 2 +-
.../typed_banchmark_producing_consumer.rs | 128 +++++
core/bench/src/benchmarks/common.rs | 42 +-
core/bench/src/utils/batch_generator.rs | 4 +
22 files changed, 1232 insertions(+), 994 deletions(-)
diff --git a/core/bench/src/actors/consumer/backend.rs b/core/bench/src/actors/consumer/backend.rs
deleted file mode 100644
index b61aaa41..00000000
--- a/core/bench/src/actors/consumer/backend.rs
+++ /dev/null
@@ -1,84 +0,0 @@
-/* Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-use bench_report::numeric_parameter::BenchmarkNumericParameter;
-use iggy::prelude::*;
-use integration::test_server::ClientFactory;
-use std::marker::PhantomData;
-use std::sync::Arc;
-use std::time::Duration;
-
-#[derive(Clone)]
-pub struct ConsumerBackendImpl<T> {
- pub client_factory: Arc<dyn ClientFactory>,
- pub config: BenchmarkConsumerConfig,
- _phantom: PhantomData<T>,
-}
-
-impl<T> ConsumerBackendImpl<T> {
- pub fn new(client_factory: Arc<dyn ClientFactory>, config: BenchmarkConsumerConfig) -> Self {
- Self {
- client_factory,
- config,
- _phantom: PhantomData,
- }
- }
-}
-
-pub struct LowLevelApiMarker;
-pub struct HighLevelApiMarker;
-
-pub type LowLevelBackend = ConsumerBackendImpl<LowLevelApiMarker>;
-pub type HighLevelBackend = ConsumerBackendImpl<HighLevelApiMarker>;
-
-pub enum ConsumerBackend {
- LowLevel(LowLevelBackend),
- HighLevel(HighLevelBackend),
-}
-
-#[derive(Debug, Clone)]
-pub struct ConsumedBatch {
- pub messages: u32,
- pub user_data_bytes: u64,
- pub total_bytes: u64,
- pub latency: Duration,
-}
-
-#[derive(Debug, Clone)]
-pub struct BenchmarkConsumerConfig {
- pub consumer_id: u32,
- pub consumer_group_id: Option<u32>,
- pub stream_id: u32,
- pub messages_per_batch: BenchmarkNumericParameter,
- pub warmup_time: IggyDuration,
- pub polling_kind: PollingKind,
- pub origin_timestamp_latency_calculation: bool,
-}
-
-pub trait BenchmarkConsumerBackend {
- type Consumer;
-
- async fn setup(&self) -> Result<Self::Consumer, IggyError>;
- async fn warmup(&self, consumer: &mut Self::Consumer) -> Result<(), IggyError>;
- async fn consume_batch(
- &self,
- consumer: &mut Self::Consumer,
- ) -> Result<Option<ConsumedBatch>, IggyError>;
- fn log_setup_info(&self);
- fn log_warmup_info(&self);
-}
diff --git a/core/bench/src/actors/consumer/benchmark_consumer.rs b/core/bench/src/actors/consumer/benchmark_consumer.rs
index b739a00d..6181fc49 100644
--- a/core/bench/src/actors/consumer/benchmark_consumer.rs
+++ b/core/bench/src/actors/consumer/benchmark_consumer.rs
@@ -16,10 +16,8 @@
* under the License.
*/
-use super::backend::{
- BenchmarkConsumerBackend, BenchmarkConsumerConfig, ConsumerBackend, HighLevelBackend,
- LowLevelBackend,
-};
+use crate::actors::consumer::client::BenchmarkConsumerClient;
+use crate::actors::consumer::client::interface::BenchmarkConsumerConfig;
use crate::analytics::metrics::individual::from_records;
use crate::analytics::record::BenchmarkRecord;
use crate::utils::finish_condition::BenchmarkFinishCondition;
@@ -30,58 +28,33 @@ use bench_report::individual_metrics::BenchmarkIndividualMetrics;
use bench_report::numeric_parameter::BenchmarkNumericParameter;
use human_repr::HumanCount;
use iggy::prelude::*;
-use integration::test_server::ClientFactory;
use std::sync::Arc;
use std::time::Duration;
use tokio::time::Instant;
use tracing::info;
-pub struct BenchmarkConsumer {
- backend: ConsumerBackend,
- benchmark_kind: BenchmarkKind,
- finish_condition: Arc<BenchmarkFinishCondition>,
- sampling_time: IggyDuration,
- moving_average_window: u32,
- limit_bytes_per_second: Option<IggyByteSize>,
- config: BenchmarkConsumerConfig,
+pub struct BenchmarkConsumer<C: BenchmarkConsumerClient> {
+ pub client: C,
+ pub benchmark_kind: BenchmarkKind,
+ pub finish_condition: Arc<BenchmarkFinishCondition>,
+ pub sampling_time: IggyDuration,
+ pub moving_average_window: u32,
+ pub limit_bytes_per_second: Option<IggyByteSize>,
+ pub config: BenchmarkConsumerConfig,
}
-impl BenchmarkConsumer {
- #[allow(clippy::too_many_arguments)]
- pub fn new(
- client_factory: Arc<dyn ClientFactory>,
+impl<C: BenchmarkConsumerClient> BenchmarkConsumer<C> {
+ pub const fn new(
+ client: C,
benchmark_kind: BenchmarkKind,
- consumer_id: u32,
- consumer_group_id: Option<u32>,
- stream_id: u32,
- messages_per_batch: BenchmarkNumericParameter,
finish_condition: Arc<BenchmarkFinishCondition>,
- warmup_time: IggyDuration,
sampling_time: IggyDuration,
moving_average_window: u32,
- polling_kind: PollingKind,
limit_bytes_per_second: Option<IggyByteSize>,
- origin_timestamp_latency_calculation: bool,
- use_high_level_api: bool,
+ config: BenchmarkConsumerConfig,
) -> Self {
- let config = BenchmarkConsumerConfig {
- consumer_id,
- consumer_group_id,
- stream_id,
- messages_per_batch,
- warmup_time,
- polling_kind,
- origin_timestamp_latency_calculation,
- };
-
- let backend = if use_high_level_api {
- ConsumerBackend::HighLevel(HighLevelBackend::new(client_factory, config.clone()))
- } else {
- ConsumerBackend::LowLevel(LowLevelBackend::new(client_factory, config.clone()))
- };
-
Self {
- backend,
+ client,
benchmark_kind,
finish_condition,
sampling_time,
@@ -91,64 +64,32 @@ impl BenchmarkConsumer {
}
}
- pub async fn run(self) -> Result<BenchmarkIndividualMetrics, IggyError> {
- match self.backend {
- ConsumerBackend::LowLevel(backend) => {
- Self::run_with_backend(
- self.benchmark_kind,
- self.finish_condition,
- self.sampling_time,
- self.moving_average_window,
- self.limit_bytes_per_second,
- self.config,
- backend,
- )
- .await
- }
- ConsumerBackend::HighLevel(backend) => {
- Self::run_with_backend(
- self.benchmark_kind,
- self.finish_condition,
- self.sampling_time,
- self.moving_average_window,
- self.limit_bytes_per_second,
- self.config,
- backend,
- )
- .await
- }
- }
- }
-
- async fn run_with_backend<B: BenchmarkConsumerBackend>(
- benchmark_kind: BenchmarkKind,
- finish_condition: Arc<BenchmarkFinishCondition>,
- sampling_time: IggyDuration,
- moving_average_window: u32,
- limit_bytes_per_second: Option<IggyByteSize>,
- config: BenchmarkConsumerConfig,
- backend: B,
- ) -> Result<BenchmarkIndividualMetrics, IggyError> {
- let mut consumer = backend.setup().await?;
+ #[allow(clippy::too_many_lines)]
+ #[allow(clippy::cognitive_complexity)]
+ pub async fn run(mut self) -> Result<BenchmarkIndividualMetrics, IggyError> {
+ self.client.setup().await?;
- if config.warmup_time.get_duration() != Duration::from_millis(0) {
- backend.log_warmup_info();
- backend.warmup(&mut consumer).await?;
+ if self.config.warmup_time.get_duration() != Duration::from_millis(0) {
+ self.log_warmup_info();
+ let warmup_end = Instant::now() + self.config.warmup_time.get_duration();
+ while Instant::now() < warmup_end {
+ let _ = self.client.consume_batch().await?;
+ }
}
- backend.log_setup_info();
+ self.log_setup_info();
- let max_capacity = finish_condition.max_capacity();
+ let max_capacity = self.finish_condition.max_capacity();
let mut records = Vec::with_capacity(max_capacity);
let mut messages_processed = 0;
let mut batches_processed = 0;
let mut bytes_processed = 0;
let mut user_data_bytes_processed = 0;
let start_timestamp = Instant::now();
- let rate_limiter = limit_bytes_per_second.map(BenchmarkRateLimiter::new);
+ let rate_limiter = self.limit_bytes_per_second.map(BenchmarkRateLimiter::new);
- while !finish_condition.is_done() {
- let batch_opt = backend.consume_batch(&mut consumer).await?;
+ while !self.finish_condition.is_done() {
+ let batch_opt = self.client.consume_batch().await?;
let Some(batch) = batch_opt else {
continue;
@@ -175,31 +116,69 @@ impl BenchmarkConsumer {
.await;
}
- if finish_condition.account_and_check(batch.user_data_bytes) {
+ if self
+ .finish_condition
+ .account_and_check(batch.user_data_bytes)
+ {
break;
}
}
let metrics = from_records(
&records,
- benchmark_kind,
+ self.benchmark_kind,
ActorKind::Consumer,
- config.consumer_id,
- sampling_time,
- moving_average_window,
+ self.config.consumer_id,
+ self.sampling_time,
+ self.moving_average_window,
);
Self::log_statistics(
- config.consumer_id,
+ self.config.consumer_id,
messages_processed,
u32::try_from(batches_processed).unwrap_or(u32::MAX),
- &config.messages_per_batch,
+ &self.config.messages_per_batch,
&metrics,
);
Ok(metrics)
}
+ pub fn log_setup_info(&self) {
+ if let Some(cg_id) = self.config.consumer_group_id {
+ info!(
+ "Consumer #{}, part of consumer group #{} → polling in {} messages per batch from stream {}, using {}...",
+ self.config.consumer_id,
+ cg_id,
+ self.config.messages_per_batch,
+ self.config.stream_id,
+ C::API_LABEL,
+ );
+ } else {
+ info!(
+ "Consumer #{} → polling in {} messages per batch from stream {}, using {}...",
+ self.config.consumer_id,
+ self.config.messages_per_batch,
+ self.config.stream_id,
+ C::API_LABEL,
+ );
+ }
+ }
+
+ pub fn log_warmup_info(&self) {
+ if let Some(cg_id) = self.config.consumer_group_id {
+ info!(
+ "Consumer #{}, group #{} → warming up for {}",
+ self.config.consumer_id, cg_id, self.config.warmup_time,
+ );
+ } else {
+ info!(
+ "Consumer #{} → warming up for {}",
+ self.config.consumer_id, self.config.warmup_time,
+ );
+ }
+ }
+
pub fn log_statistics(
consumer_id: u32,
total_messages: u64,
diff --git a/core/bench/src/actors/consumer/high_level_backend.rs b/core/bench/src/actors/consumer/client/high_level.rs
similarity index 55%
rename from core/bench/src/actors/consumer/high_level_backend.rs
rename to core/bench/src/actors/consumer/client/high_level.rs
index 656fdace..362fd808 100644
--- a/core/bench/src/actors/consumer/high_level_backend.rs
+++ b/core/bench/src/actors/consumer/client/high_level.rs
@@ -16,80 +16,47 @@
* under the License.
*/
-use super::backend::{BenchmarkConsumerBackend, ConsumedBatch, HighLevelBackend};
+use crate::actors::{
+ ApiLabel, BatchMetrics, BenchmarkInit,
+ consumer::client::{
+ BenchmarkConsumerClient,
+ interface::{BenchmarkConsumerConfig, ConsumerClient},
+ },
+};
+
use futures_util::StreamExt;
use iggy::prelude::*;
-use integration::test_server::login_root;
-use std::time::Duration;
+use integration::test_server::{ClientFactory, login_root};
+use std::{sync::Arc, time::Duration};
use tokio::time::{Instant, timeout};
-use tracing::{debug, error, info, warn};
-
-impl BenchmarkConsumerBackend for HighLevelBackend {
- type Consumer = IggyConsumer;
-
- async fn setup(&self) -> Result<Self::Consumer, IggyError> {
- let topic_id: u32 = 1;
- let client = self.client_factory.create_client().await;
- let client = IggyClient::create(client, None, None);
- login_root(&client).await;
-
- let stream_id_str = self.config.stream_id.to_string();
- let topic_id_str = topic_id.to_string();
-
- let mut iggy_consumer = if let Some(cg_id) = self.config.consumer_group_id {
- let consumer_group_name = format!("cg_{cg_id}");
- // Consumer groups use auto-commit (matching PollingKind::Next behavior from low-level API)
- client
- .consumer_group(&consumer_group_name, &stream_id_str, &topic_id_str)?
- .batch_length(self.config.messages_per_batch.get())
- .auto_commit(AutoCommit::When(AutoCommitWhen::PollingMessages))
- .create_consumer_group_if_not_exists()
- .auto_join_consumer_group()
- .build()
- } else {
- // TODO(hubcio): as of now, there is no way to mimic the behavior of
- // PollingKind::Offset, because high level API doesn't provide method
- // to commit local offset manually, only auto-commit on server.
- client
- .consumer(
- &format!("hl_consumer_{}", self.config.consumer_id),
- &stream_id_str,
- &topic_id_str,
- 1,
- )?
- .batch_length(self.config.messages_per_batch.get())
- .auto_commit(AutoCommit::Disabled)
- .build()
- };
+use tracing::{error, warn};
- iggy_consumer.init().await?;
- Ok(iggy_consumer)
- }
+pub struct HighLevelConsumerClient {
+ client_factory: Arc<dyn ClientFactory>,
+ config: BenchmarkConsumerConfig,
+ consumer: Option<IggyConsumer>,
+}
- async fn warmup(&self, consumer: &mut Self::Consumer) -> Result<(), IggyError> {
- let warmup_end = Instant::now() + self.config.warmup_time.get_duration();
- while Instant::now() < warmup_end {
- if let Some(message) = consumer.next().await {
- if message.is_err() {
- break;
- }
- }
+impl HighLevelConsumerClient {
+ pub fn new(client_factory: Arc<dyn ClientFactory>, config: BenchmarkConsumerConfig) -> Self {
+ Self {
+ client_factory,
+ config,
+ consumer: None,
}
- Ok(())
}
+}
+
+impl ConsumerClient for HighLevelConsumerClient {
+ async fn consume_batch(&mut self) -> Result<Option<BatchMetrics>, IggyError> {
+ let consumer = self.consumer.as_mut().expect("Consumer not initialized");
- #[allow(clippy::cognitive_complexity)]
- async fn consume_batch(
- &self,
- consumer: &mut Self::Consumer,
- ) -> Result<Option<ConsumedBatch>, IggyError> {
let batch_start = Instant::now();
let mut batch_messages = 0;
let mut batch_user_bytes = 0;
let mut batch_total_bytes = 0;
while batch_messages < self.config.messages_per_batch.get() {
- // Use timeout to avoid getting stuck waiting for messages
let timeout_result = timeout(Duration::from_secs(1), consumer.next()).await;
match timeout_result {
@@ -101,19 +68,11 @@ impl BenchmarkConsumerBackend for HighLevelBackend {
received_message.message.get_size_bytes().as_bytes_u64();
let offset = received_message.message.header.offset;
- if batch_messages >= self.config.messages_per_batch.get() {
- info!(
- "Batch of {} messages consumed, last_offset: {}, current_offset: {}",
- batch_messages,
- received_message.message.header.offset,
- received_message.current_offset
- );
+ if batch_messages >= self.config.messages_per_batch.get() {
if let Err(error) = consumer.store_offset(offset, None).await {
error!("Failed to store offset: {offset}. {error}");
- continue;
}
- debug!("Offset: {offset} stored successfully");
break;
}
}
@@ -121,15 +80,7 @@ impl BenchmarkConsumerBackend for HighLevelBackend {
warn!("Error receiving message: {}", err);
}
},
- Ok(None) => {
- debug!("Consumer stream ended during batching");
- break;
- }
- Err(_) => {
- debug!(
- "Timeout waiting for messages, stopping batch at {} messages",
- batch_messages
- );
+ Ok(None) | Err(_) => {
break;
}
}
@@ -138,7 +89,7 @@ impl BenchmarkConsumerBackend for HighLevelBackend {
if batch_messages == 0 {
Ok(None)
} else {
- Ok(Some(ConsumedBatch {
+ Ok(Some(BatchMetrics {
messages: batch_messages,
user_data_bytes: batch_user_bytes,
total_bytes: batch_total_bytes,
@@ -146,35 +97,52 @@ impl BenchmarkConsumerBackend for HighLevelBackend {
}))
}
}
+}
- fn log_setup_info(&self) {
- if let Some(cg_id) = self.config.consumer_group_id {
- info!(
- "Consumer #{}, part of consumer group #{} → polling in {} messages per batch from stream {}, using high-level API...",
- self.config.consumer_id,
- cg_id,
- self.config.messages_per_batch,
- self.config.stream_id,
- );
- } else {
- info!(
- "Consumer #{} → polling in {} messages per batch from stream {}, using high-level API...",
- self.config.consumer_id, self.config.messages_per_batch, self.config.stream_id,
- );
- }
- }
+impl BenchmarkInit for HighLevelConsumerClient {
+ async fn setup(&mut self) -> Result<(), IggyError> {
+ let topic_id = 1;
+ let client = self.client_factory.create_client().await;
+ let client = IggyClient::create(client, None, None);
+ login_root(&client).await;
- fn log_warmup_info(&self) {
- if let Some(cg_id) = self.config.consumer_group_id {
- info!(
- "Consumer #{}, part of consumer group #{}, → warming up for {}...",
- self.config.consumer_id, cg_id, self.config.warmup_time
- );
+ let stream_id_str = self.config.stream_id.to_string();
+ let topic_id_str = topic_id.to_string();
+
+ let mut consumer = if let Some(cg_id) = self.config.consumer_group_id {
+ let consumer_group_name = format!("cg_{cg_id}");
+ client
+ .consumer_group(&consumer_group_name, &stream_id_str, &topic_id_str)?
+ .batch_length(self.config.messages_per_batch.get())
+ .auto_commit(AutoCommit::When(AutoCommitWhen::PollingMessages))
+ .create_consumer_group_if_not_exists()
+ .auto_join_consumer_group()
+ .build()
} else {
- info!(
- "Consumer #{} → warming up for {}...",
- self.config.consumer_id, self.config.warmup_time
- );
- }
+ // TODO(hubcio): as of now, there is no way to mimic the behavior of
+ // PollingKind::Offset, because high level API doesn't provide method
+ // to commit local offset manually, only auto-commit on server.
+ client
+ .consumer(
+ &format!("hl_consumer_{}", self.config.consumer_id),
+ &stream_id_str,
+ &topic_id_str,
+ 1,
+ )?
+ .polling_strategy(PollingStrategy::offset(0))
+ .batch_length(self.config.messages_per_batch.get())
+ .auto_commit(AutoCommit::Disabled)
+ .build()
+ };
+
+ consumer.init().await?;
+ self.consumer = Some(consumer);
+ Ok(())
}
}
+
+impl ApiLabel for HighLevelConsumerClient {
+ const API_LABEL: &'static str = "high-level";
+}
+
+impl BenchmarkConsumerClient for HighLevelConsumerClient {}
diff --git a/core/bench/src/actors/producing_consumer/mod.rs b/core/bench/src/actors/consumer/client/interface.rs
similarity index 53%
copy from core/bench/src/actors/producing_consumer/mod.rs
copy to core/bench/src/actors/consumer/client/interface.rs
index 2927c8d9..d1179a73 100644
--- a/core/bench/src/actors/producing_consumer/mod.rs
+++ b/core/bench/src/actors/consumer/client/interface.rs
@@ -16,6 +16,23 @@
* under the License.
*/
-pub mod benchmark_producing_consumer;
+use bench_report::numeric_parameter::BenchmarkNumericParameter;
+use iggy::prelude::*;
-pub use benchmark_producing_consumer::BenchmarkProducingConsumer;
+use crate::actors::{ApiLabel, BatchMetrics, BenchmarkInit};
+
+#[derive(Debug, Clone)]
+pub struct BenchmarkConsumerConfig {
+ pub consumer_id: u32,
+ pub consumer_group_id: Option<u32>,
+ pub stream_id: u32,
+ pub messages_per_batch: BenchmarkNumericParameter,
+ pub warmup_time: IggyDuration,
+ pub polling_kind: PollingKind,
+ pub origin_timestamp_latency_calculation: bool,
+}
+
+pub trait ConsumerClient: Send + Sync {
+ async fn consume_batch(&mut self) -> Result<Option<BatchMetrics>, IggyError>;
+}
+pub trait BenchmarkConsumerClient: ConsumerClient + BenchmarkInit + ApiLabel + Send + Sync {}
diff --git a/core/bench/src/actors/consumer/client/low_level.rs b/core/bench/src/actors/consumer/client/low_level.rs
new file mode 100644
index 00000000..3217a603
--- /dev/null
+++ b/core/bench/src/actors/consumer/client/low_level.rs
@@ -0,0 +1,162 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use crate::actors::consumer::client::BenchmarkConsumerClient;
+use crate::actors::consumer::client::interface::{BenchmarkConsumerConfig, ConsumerClient};
+use crate::actors::{ApiLabel, BatchMetrics, BenchmarkInit};
+use crate::benchmarks::common::create_consumer;
+use crate::utils::{batch_total_size_bytes, batch_user_size_bytes};
+use iggy::prelude::*;
+use integration::test_server::{ClientFactory, login_root};
+use std::sync::Arc;
+use std::time::Duration;
+use tokio::time::Instant;
+
+pub struct LowLevelConsumerClient {
+ client_factory: Arc<dyn ClientFactory>,
+ config: BenchmarkConsumerConfig,
+ client: Option<IggyClient>,
+ consumer: Option<Consumer>,
+ stream_id: Identifier,
+ topic_id: Identifier,
+ partition_id: Option<u32>,
+ polling_strategy: PollingStrategy,
+ auto_commit: bool,
+ offset: u64,
+}
+
+impl LowLevelConsumerClient {
+ pub fn new(client_factory: Arc<dyn ClientFactory>, config: BenchmarkConsumerConfig) -> Self {
+ Self {
+ client_factory,
+ config,
+ client: None,
+ consumer: None,
+ stream_id: Identifier::default(),
+ topic_id: Identifier::default(),
+ partition_id: None,
+ polling_strategy: PollingStrategy::next(),
+ auto_commit: true,
+ offset: 0,
+ }
+ }
+}
+
+impl ConsumerClient for LowLevelConsumerClient {
+ async fn consume_batch(&mut self) -> Result<Option<BatchMetrics>, IggyError> {
+ let client = self.client.as_ref().expect("client not initialized");
+ let consumer = self.consumer.as_ref().expect("consumer not initialized");
+ let messages_to_receive = self.config.messages_per_batch.get();
+
+ let before_poll = Instant::now();
+ let polled = client
+ .poll_messages(
+ &self.stream_id,
+ &self.topic_id,
+ self.partition_id,
+ consumer,
+ &self.polling_strategy,
+ messages_to_receive,
+ self.auto_commit,
+ )
+ .await;
+
+ let polled = match polled {
+ Ok(p) => p,
+ Err(e) => {
+ if matches!(e, IggyError::TopicIdNotFound(_, _)) {
+ return Ok(None);
+ }
+ return Err(e);
+ }
+ };
+
+ if polled.messages.is_empty() {
+ return Ok(None);
+ }
+ let message_count = u32::try_from(polled.messages.len()).unwrap();
+ let latency = if self.config.origin_timestamp_latency_calculation {
+ let now = IggyTimestamp::now().as_micros();
+ Duration::from_micros(now - polled.messages[0].header.origin_timestamp)
+ } else {
+ before_poll.elapsed()
+ };
+
+ let user_bytes = batch_user_size_bytes(&polled);
+ let total_bytes = batch_total_size_bytes(&polled);
+
+ self.offset += polled.messages.len() as u64;
+
+ Ok(Some(BatchMetrics {
+ messages: message_count,
+ user_data_bytes: user_bytes,
+ total_bytes,
+ latency,
+ }))
+ }
+}
+
+impl BenchmarkInit for LowLevelConsumerClient {
+ async fn setup(&mut self) -> Result<(), IggyError> {
+ let topic_id = 1u32;
+ let default_partition_id = 1u32;
+
+ let client = self.client_factory.create_client().await;
+ let client = IggyClient::create(client, None, None);
+ login_root(&client).await;
+
+ let stream_id = self.config.stream_id.try_into().unwrap();
+ let topic_id = topic_id.try_into().unwrap();
+ let partition_id = if self.config.consumer_group_id.is_some() {
+ None
+ } else {
+ Some(default_partition_id)
+ };
+
+ let consumer = create_consumer(
+ &client,
+ self.config.consumer_group_id.as_ref(),
+ &stream_id,
+ &topic_id,
+ self.config.consumer_id,
+ )
+ .await;
+ let (polling_strategy, auto_commit) = match self.config.polling_kind {
+ PollingKind::Offset => (PollingStrategy::offset(self.offset), false),
+ PollingKind::Next => (PollingStrategy::next(), true),
+ _ => panic!("Unsupported polling kind: {:?}", self.config.polling_kind),
+ };
+
+ self.client = Some(client);
+ self.consumer = Some(consumer);
+
+ self.stream_id = stream_id;
+ self.topic_id = topic_id;
+ self.partition_id = partition_id;
+ self.polling_strategy = polling_strategy;
+ self.auto_commit = auto_commit;
+
+ Ok(())
+ }
+}
+
+impl ApiLabel for LowLevelConsumerClient {
+ const API_LABEL: &'static str = "low-level";
+}
+
+impl BenchmarkConsumerClient for LowLevelConsumerClient {}
diff --git a/core/bench/src/actors/producer/mod.rs b/core/bench/src/actors/consumer/client/mod.rs
similarity index 88%
copy from core/bench/src/actors/producer/mod.rs
copy to core/bench/src/actors/consumer/client/mod.rs
index d4f6209f..7d17c685 100644
--- a/core/bench/src/actors/producer/mod.rs
+++ b/core/bench/src/actors/consumer/client/mod.rs
@@ -16,6 +16,8 @@
* under the License.
*/
-pub mod benchmark_producer;
+pub mod high_level;
+pub mod interface;
+pub mod low_level;
-pub use benchmark_producer::BenchmarkProducer;
+pub use interface::BenchmarkConsumerClient;
diff --git a/core/bench/src/actors/consumer/low_level_backend.rs b/core/bench/src/actors/consumer/low_level_backend.rs
deleted file mode 100644
index bcdb3f3e..00000000
--- a/core/bench/src/actors/consumer/low_level_backend.rs
+++ /dev/null
@@ -1,201 +0,0 @@
-/* Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-use super::backend::{BenchmarkConsumerBackend, ConsumedBatch, LowLevelBackend};
-use crate::benchmarks::common::create_consumer;
-use crate::utils::{batch_total_size_bytes, batch_user_size_bytes};
-use iggy::prelude::*;
-use integration::test_server::login_root;
-use std::time::Duration;
-use tokio::time::Instant;
-use tracing::{info, warn};
-
-impl BenchmarkConsumerBackend for LowLevelBackend {
- type Consumer = (
- IggyClient,
- Identifier,
- Identifier,
- Option<u32>,
- Consumer,
- u64,
- );
-
- async fn setup(&self) -> Result<Self::Consumer, IggyError> {
- let topic_id: u32 = 1;
- let default_partition_id: u32 = 1;
- let client = self.client_factory.create_client().await;
- let client = IggyClient::create(client, None, None);
- login_root(&client).await;
-
- let stream_id = self.config.stream_id.try_into().unwrap();
- let topic_id = topic_id.try_into().unwrap();
- let partition_id = if self.config.consumer_group_id.is_some() {
- None
- } else {
- Some(default_partition_id)
- };
- let cg_id = self.config.consumer_group_id;
- let consumer = create_consumer(
- &client,
- cg_id.as_ref(),
- &stream_id,
- &topic_id,
- self.config.consumer_id,
- )
- .await;
-
- Ok((client, stream_id, topic_id, partition_id, consumer, 0))
- }
-
- async fn warmup(&self, consumer: &mut Self::Consumer) -> Result<(), IggyError> {
- let (client, stream_id, topic_id, partition_id, consumer_obj, messages_processed) =
- consumer;
- let warmup_end = Instant::now() + self.config.warmup_time.get_duration();
- let mut warmup_messages_processed = 0;
- let mut _last_batch_user_size_bytes = 0;
-
- while Instant::now() < warmup_end {
- let messages_to_receive = self.config.messages_per_batch.get();
- let offset = *messages_processed + warmup_messages_processed;
- let (strategy, auto_commit) = match self.config.polling_kind {
- PollingKind::Offset => (PollingStrategy::offset(offset), false),
- PollingKind::Next => (PollingStrategy::next(), true),
- _ => panic!(
- "Unsupported polling kind for benchmark: {:?}",
- self.config.polling_kind
- ),
- };
- let polled_messages = client
- .poll_messages(
- stream_id,
- topic_id,
- *partition_id,
- consumer_obj,
- &strategy,
- messages_to_receive,
- auto_commit,
- )
- .await?;
-
- if polled_messages.messages.is_empty() {
- warn!(
- "Consumer #{} → Messages are empty for offset: {}, retrying...",
- self.config.consumer_id, offset
- );
- continue;
- }
- warmup_messages_processed += polled_messages.messages.len() as u64;
- _last_batch_user_size_bytes = batch_user_size_bytes(&polled_messages);
- }
- Ok(())
- }
-
- async fn consume_batch(
- &self,
- consumer: &mut Self::Consumer,
- ) -> Result<Option<ConsumedBatch>, IggyError> {
- let (client, stream_id, topic_id, partition_id, consumer_obj, messages_processed) =
- consumer;
- let messages_to_receive = self.config.messages_per_batch.get();
- let offset = *messages_processed;
- let (strategy, auto_commit) = match self.config.polling_kind {
- PollingKind::Offset => (PollingStrategy::offset(offset), false),
- PollingKind::Next => (PollingStrategy::next(), true),
- _ => panic!(
- "Unsupported polling kind for benchmark: {:?}",
- self.config.polling_kind
- ),
- };
-
- let before_poll = Instant::now();
- let polled_messages = client
- .poll_messages(
- stream_id,
- topic_id,
- *partition_id,
- consumer_obj,
- &strategy,
- messages_to_receive,
- auto_commit,
- )
- .await;
-
- let polled_messages = match polled_messages {
- Ok(messages) => messages,
- Err(e) => {
- if matches!(e, IggyError::TopicIdNotFound(_, _)) {
- return Ok(None);
- }
- return Err(e);
- }
- };
-
- if polled_messages.messages.is_empty() {
- return Ok(None);
- }
- let latency = if self.config.origin_timestamp_latency_calculation {
- let now = IggyTimestamp::now().as_micros();
- Duration::from_micros(now - polled_messages.messages[0].header.origin_timestamp)
- } else {
- before_poll.elapsed()
- };
-
- let batch_user_size_bytes = batch_user_size_bytes(&polled_messages);
- let batch_total_size_bytes = batch_total_size_bytes(&polled_messages);
-
- *messages_processed += polled_messages.messages.len() as u64;
-
- Ok(Some(ConsumedBatch {
- messages: u32::try_from(polled_messages.messages.len()).unwrap(),
- user_data_bytes: batch_user_size_bytes,
- total_bytes: batch_total_size_bytes,
- latency,
- }))
- }
-
- fn log_setup_info(&self) {
- if let Some(cg_id) = self.config.consumer_group_id {
- info!(
- "Consumer #{}, part of consumer group #{} → polling in {} messages per batch from stream {}, using low-level API...",
- self.config.consumer_id,
- cg_id,
- self.config.messages_per_batch,
- self.config.stream_id,
- );
- } else {
- info!(
- "Consumer #{} → polling in {} messages per batch from stream {}, using low-level API...",
- self.config.consumer_id, self.config.messages_per_batch, self.config.stream_id,
- );
- }
- }
-
- fn log_warmup_info(&self) {
- if let Some(cg_id) = self.config.consumer_group_id {
- info!(
- "Consumer #{}, part of consumer group #{}, → warming up for {}...",
- self.config.consumer_id, cg_id, self.config.warmup_time
- );
- } else {
- info!(
- "Consumer #{} → warming up for {}...",
- self.config.consumer_id, self.config.warmup_time
- );
- }
- }
-}
diff --git a/core/bench/src/actors/consumer/mod.rs b/core/bench/src/actors/consumer/mod.rs
index 2912806b..c7e2ae46 100644
--- a/core/bench/src/actors/consumer/mod.rs
+++ b/core/bench/src/actors/consumer/mod.rs
@@ -16,9 +16,7 @@
* under the License.
*/
-pub mod backend;
pub mod benchmark_consumer;
-pub mod high_level_backend;
-pub mod low_level_backend;
-
+pub mod client;
+pub mod typed_benchmark_consumer;
pub use benchmark_consumer::BenchmarkConsumer;
diff --git a/core/bench/src/actors/consumer/typed_benchmark_consumer.rs b/core/bench/src/actors/consumer/typed_benchmark_consumer.rs
new file mode 100644
index 00000000..fca492ea
--- /dev/null
+++ b/core/bench/src/actors/consumer/typed_benchmark_consumer.rs
@@ -0,0 +1,100 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use std::sync::Arc;
+
+use crate::{
+ actors::consumer::{
+ BenchmarkConsumer,
+ client::{
+ high_level::HighLevelConsumerClient, interface::BenchmarkConsumerConfig,
+ low_level::LowLevelConsumerClient,
+ },
+ },
+ utils::finish_condition::BenchmarkFinishCondition,
+};
+use bench_report::{
+ benchmark_kind::BenchmarkKind, individual_metrics::BenchmarkIndividualMetrics,
+ numeric_parameter::BenchmarkNumericParameter,
+};
+use iggy::prelude::*;
+use integration::test_server::ClientFactory;
+
+pub enum TypedBenchmarkConsumer {
+ High(BenchmarkConsumer<HighLevelConsumerClient>),
+ Low(BenchmarkConsumer<LowLevelConsumerClient>),
+}
+
+impl TypedBenchmarkConsumer {
+ #[allow(clippy::too_many_arguments)]
+ pub fn new(
+ use_high_level_api: bool,
+ client_factory: Arc<dyn ClientFactory>,
+ benchmark_kind: BenchmarkKind,
+ consumer_id: u32,
+ consumer_group_id: Option<u32>,
+ stream_id: u32,
+ messages_per_batch: BenchmarkNumericParameter,
+ finish_condition: Arc<BenchmarkFinishCondition>,
+ warmup_time: IggyDuration,
+ sampling_time: IggyDuration,
+ moving_average_window: u32,
+ polling_kind: PollingKind,
+ limit_bytes_per_second: Option<IggyByteSize>,
+ origin_timestamp_latency_calculation: bool,
+ ) -> Self {
+ let config = BenchmarkConsumerConfig {
+ consumer_id,
+ consumer_group_id,
+ stream_id,
+ messages_per_batch,
+ warmup_time,
+ polling_kind,
+ origin_timestamp_latency_calculation,
+ };
+
+ if use_high_level_api {
+ Self::High(BenchmarkConsumer::new(
+ HighLevelConsumerClient::new(client_factory, config.clone()),
+ benchmark_kind,
+ finish_condition,
+ sampling_time,
+ moving_average_window,
+ limit_bytes_per_second,
+ config,
+ ))
+ } else {
+ Self::Low(BenchmarkConsumer::new(
+ LowLevelConsumerClient::new(client_factory, config.clone()),
+ benchmark_kind,
+ finish_condition,
+ sampling_time,
+ moving_average_window,
+ limit_bytes_per_second,
+ config,
+ ))
+ }
+ }
+
+ pub async fn run(self) -> Result<BenchmarkIndividualMetrics, IggyError> {
+ match self {
+ Self::High(c) => c.run().await,
+ Self::Low(c) => c.run().await,
+ }
+ }
+}
diff --git a/core/bench/src/actors/mod.rs b/core/bench/src/actors/mod.rs
index 128fafe1..d992ad5a 100644
--- a/core/bench/src/actors/mod.rs
+++ b/core/bench/src/actors/mod.rs
@@ -16,6 +16,26 @@
* under the License.
*/
+use std::time::Duration;
+
+use iggy::prelude::IggyError;
+
pub mod consumer;
pub mod producer;
pub mod producing_consumer;
+
+#[derive(Debug, Clone)]
+pub struct BatchMetrics {
+ pub messages: u32,
+ pub user_data_bytes: u64,
+ pub total_bytes: u64,
+ pub latency: Duration,
+}
+
+pub trait BenchmarkInit: Send + Sync {
+ async fn setup(&mut self) -> Result<(), IggyError>;
+}
+
+pub trait ApiLabel {
+ const API_LABEL: &'static str;
+}
diff --git a/core/bench/src/actors/producer/benchmark_producer.rs b/core/bench/src/actors/producer/benchmark_producer.rs
index dff7b18b..9c9f2a55 100644
--- a/core/bench/src/actors/producer/benchmark_producer.rs
+++ b/core/bench/src/actors/producer/benchmark_producer.rs
@@ -16,144 +16,102 @@
* under the License.
*/
-use crate::analytics::metrics::individual::from_records;
-use crate::analytics::record::BenchmarkRecord;
-use crate::utils::batch_generator::BenchmarkBatchGenerator;
-use crate::utils::finish_condition::BenchmarkFinishCondition;
-use crate::utils::rate_limiter::BenchmarkRateLimiter;
+use crate::{
+ actors::producer::client::{BenchmarkProducerClient, interface::BenchmarkProducerConfig},
+ analytics::{metrics::individual::from_records, record::BenchmarkRecord},
+ utils::{
+ batch_generator::BenchmarkBatchGenerator, finish_condition::BenchmarkFinishCondition,
+ rate_limiter::BenchmarkRateLimiter,
+ },
+};
use bench_report::actor_kind::ActorKind;
use bench_report::benchmark_kind::BenchmarkKind;
use bench_report::individual_metrics::BenchmarkIndividualMetrics;
use bench_report::numeric_parameter::BenchmarkNumericParameter;
use human_repr::HumanCount;
use iggy::prelude::*;
-use integration::test_server::{ClientFactory, login_root};
-use std::sync::Arc;
-use std::time::Duration;
+use std::{sync::Arc, time::Duration};
use tokio::time::Instant;
use tracing::info;
-pub struct BenchmarkProducer {
- client_factory: Arc<dyn ClientFactory>,
- benchmark_kind: BenchmarkKind,
- producer_id: u32,
- stream_id: u32,
- partitions: u32,
- messages_per_batch: BenchmarkNumericParameter,
- message_size: BenchmarkNumericParameter,
- finish_condition: Arc<BenchmarkFinishCondition>,
- warmup_time: IggyDuration,
- sampling_time: IggyDuration,
- moving_average_window: u32,
- limit_bytes_per_second: Option<IggyByteSize>,
+pub struct BenchmarkProducer<P: BenchmarkProducerClient> {
+ pub client: P,
+ pub benchmark_kind: BenchmarkKind,
+ pub finish_condition: Arc<BenchmarkFinishCondition>,
+ pub sampling_time: IggyDuration,
+ pub moving_average_window: u32,
+ pub limit_bytes_per_second: Option<IggyByteSize>,
+ pub config: BenchmarkProducerConfig,
}
-impl BenchmarkProducer {
+impl<P: BenchmarkProducerClient> BenchmarkProducer<P> {
#[allow(clippy::too_many_arguments)]
- pub fn new(
- client_factory: Arc<dyn ClientFactory>,
+ pub const fn new(
+ client: P,
benchmark_kind: BenchmarkKind,
- producer_id: u32,
- stream_id: u32,
- partitions: u32,
- messages_per_batch: BenchmarkNumericParameter,
- message_size: BenchmarkNumericParameter,
finish_condition: Arc<BenchmarkFinishCondition>,
- warmup_time: IggyDuration,
sampling_time: IggyDuration,
moving_average_window: u32,
limit_bytes_per_second: Option<IggyByteSize>,
+ config: BenchmarkProducerConfig,
) -> Self {
Self {
- client_factory,
+ client,
benchmark_kind,
- producer_id,
- stream_id,
- partitions,
- messages_per_batch,
- message_size,
finish_condition,
- warmup_time,
sampling_time,
moving_average_window,
limit_bytes_per_second,
+ config,
}
}
+ #[allow(clippy::too_many_lines)]
#[allow(clippy::cognitive_complexity)]
- pub async fn run(self) -> Result<BenchmarkIndividualMetrics, IggyError> {
+ pub async fn run(mut self) -> Result<BenchmarkIndividualMetrics, IggyError> {
+ info!(
+ "Producer #{} → sending {} ({} msgs/batch) on stream {}, rate limit: {:?}",
+ self.config.producer_id,
+ self.finish_condition.total_str(),
+ self.config.messages_per_batch,
+ self.config.stream_id,
+ self.limit_bytes_per_second,
+ );
+
+ self.client.setup().await?;
+
let mut batch_generator =
- BenchmarkBatchGenerator::new(self.message_size, self.messages_per_batch);
- let rate_limiter = self.limit_bytes_per_second.map(BenchmarkRateLimiter::new);
+ BenchmarkBatchGenerator::new(self.config.message_size, self.config.messages_per_batch);
- let topic_id: u32 = 1;
- let default_partition_id: u32 = 1;
- let partitions = self.partitions;
- let client = self.client_factory.create_client().await;
- let client = IggyClient::create(client, None, None);
- login_root(&client).await;
-
- let stream_id = self.stream_id.try_into()?;
- let topic_id = topic_id.try_into()?;
- let partitioning = match partitions {
- 0 => panic!("Partition count must be greater than 0"),
- 1 => Partitioning::partition_id(default_partition_id),
- 2.. => Partitioning::balanced(),
- };
-
- // -----------------------
- // WARM-UP
- // -----------------------
-
- if self.warmup_time.get_duration() != Duration::from_millis(0) {
- info!(
- "Producer #{} → warming up for {}...",
- self.producer_id, self.warmup_time
- );
- let warmup_end = Instant::now() + self.warmup_time.get_duration();
+ if self.config.warmup_time.get_duration() != Duration::from_millis(0) {
+ self.log_warmup_info();
+ let warmup_end = Instant::now() + self.config.warmup_time.get_duration();
while Instant::now() < warmup_end {
- let batch = batch_generator.generate_batch();
- client
- .send_messages(&stream_id, &topic_id, &partitioning, &mut batch.messages)
- .await?;
+ let _ = self.client.produce_batch(&mut batch_generator).await?;
}
}
- // -----------------------
- // MAIN BENCHMARK
- // -----------------------
- info!(
- "Producer #{} → sending {} in batches of {} messages to stream {} with {} partitions, partitioning: {}, rate limit: {:?}...",
- self.producer_id,
- self.finish_condition.total_str(),
- self.messages_per_batch,
- stream_id,
- partitions,
- partitioning,
- self.limit_bytes_per_second
- );
+ self.log_setup_info();
let max_capacity = self.finish_condition.max_capacity();
- let mut records: Vec<BenchmarkRecord> = Vec::with_capacity(max_capacity);
+ let mut records = Vec::with_capacity(max_capacity);
let mut messages_processed = 0;
let mut batches_processed = 0;
- let mut total_bytes_processed = 0;
let mut user_data_bytes_processed = 0;
+ let mut total_bytes_processed = 0;
+
+ let rate_limiter = self.limit_bytes_per_second.map(BenchmarkRateLimiter::new);
let start_timestamp = Instant::now();
- loop {
- if self.finish_condition.is_done() {
- break;
- }
- let batch = batch_generator.generate_batch();
- let before_send = Instant::now();
- client
- .send_messages(&stream_id, &topic_id, &partitioning, &mut batch.messages)
- .await?;
- let latency = before_send.elapsed();
-
- messages_processed += batch.messages.len() as u64;
+ while !self.finish_condition.is_done() {
+ let batch_opt = self.client.produce_batch(&mut batch_generator).await?;
+
+ let Some(batch) = batch_opt else {
+ continue;
+ };
+
+ messages_processed += u64::from(batch.messages);
batches_processed += 1;
user_data_bytes_processed += batch.user_data_bytes;
total_bytes_processed += batch.total_bytes;
@@ -161,7 +119,7 @@ impl BenchmarkProducer {
records.push(BenchmarkRecord {
elapsed_time_us: u64::try_from(start_timestamp.elapsed().as_micros())
.unwrap_or(u64::MAX),
- latency_us: u64::try_from(latency.as_micros()).unwrap_or(u64::MAX),
+ latency_us: u64::try_from(batch.latency.as_micros()).unwrap_or(u64::MAX),
messages: messages_processed,
message_batches: batches_processed,
user_data_bytes: user_data_bytes_processed,
@@ -178,6 +136,15 @@ impl BenchmarkProducer {
.finish_condition
.account_and_check(batch.user_data_bytes)
{
+ info!(
+ "Producer #{} → finished sending {} messages in {} batches ({} user bytes, {} total bytes), send finish condition: {}",
+ self.config.producer_id,
+ messages_processed.human_count_bare(),
+ batches_processed.human_count_bare(),
+ user_data_bytes_processed.human_count_bytes(),
+ total_bytes_processed.human_count_bytes(),
+ self.finish_condition.status(),
+ );
break;
}
}
@@ -186,22 +153,41 @@ impl BenchmarkProducer {
&records,
self.benchmark_kind,
ActorKind::Producer,
- self.producer_id,
+ self.config.producer_id,
self.sampling_time,
self.moving_average_window,
);
Self::log_statistics(
- self.producer_id,
+ self.config.producer_id,
messages_processed,
batches_processed,
- &self.messages_per_batch,
+ &self.config.messages_per_batch,
&metrics,
);
Ok(metrics)
}
+ fn log_setup_info(&self) {
+ info!(
+ "Producer #{} → sending {} messages per batch (each {} bytes) to stream {} across {} partition(s), using {}...",
+ self.config.producer_id,
+ self.config.messages_per_batch,
+ self.config.message_size,
+ self.config.stream_id,
+ self.config.partitions,
+ P::API_LABEL,
+ );
+ }
+
+ fn log_warmup_info(&self) {
+ info!(
+ "Producer #{} → warming up for {}...",
+ self.config.producer_id, self.config.warmup_time,
+ );
+ }
+
fn log_statistics(
producer_id: u32,
total_messages: u64,
diff --git a/core/bench/src/actors/producer/client/high_level.rs b/core/bench/src/actors/producer/client/high_level.rs
new file mode 100644
index 00000000..e740716b
--- /dev/null
+++ b/core/bench/src/actors/producer/client/high_level.rs
@@ -0,0 +1,103 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use crate::actors::{
+ ApiLabel, BatchMetrics, BenchmarkInit,
+ producer::client::{
+ BenchmarkProducerClient,
+ interface::{BenchmarkProducerConfig, ProducerClient},
+ },
+};
+use iggy::prelude::*;
+use integration::test_server::{ClientFactory, login_root};
+use std::sync::Arc;
+use tokio::time::Instant;
+
+pub struct HighLevelProducerClient {
+ client_factory: Arc<dyn ClientFactory>,
+ config: BenchmarkProducerConfig,
+ producer: Option<IggyProducer>,
+}
+
+impl HighLevelProducerClient {
+ pub fn new(client_factory: Arc<dyn ClientFactory>, config: BenchmarkProducerConfig) -> Self {
+ Self {
+ client_factory,
+ config,
+ producer: None,
+ }
+ }
+}
+#[allow(clippy::significant_drop_tightening)]
+impl ProducerClient for HighLevelProducerClient {
+ async fn produce_batch(
+ &mut self,
+ batch_generator: &mut crate::utils::batch_generator::BenchmarkBatchGenerator,
+ ) -> Result<Option<BatchMetrics>, IggyError> {
+ let producer = self.producer.as_mut().expect("Producer not initialized");
+
+ let batch = batch_generator.generate_owned_batch();
+ if batch.messages.is_empty() {
+ return Ok(None);
+ }
+ let message_count = u32::try_from(batch.messages.len()).unwrap();
+ let before_send = Instant::now();
+ producer.send(batch.messages).await?;
+ let latency = before_send.elapsed();
+
+ Ok(Some(BatchMetrics {
+ messages: message_count,
+ user_data_bytes: batch.user_data_bytes,
+ total_bytes: batch.total_bytes,
+ latency,
+ }))
+ }
+}
+
+impl BenchmarkInit for HighLevelProducerClient {
+ async fn setup(&mut self) -> Result<(), IggyError> {
+ let topic_id: u32 = 1;
+
+ let client = self.client_factory.create_client().await;
+ let client = IggyClient::create(client, None, None);
+ login_root(&client).await;
+
+ let stream_id_str = self.config.stream_id.to_string();
+ let topic_id_str = topic_id.to_string();
+
+ self.producer = Some(
+ client
+ .producer(&stream_id_str, &topic_id_str)?
+ .create_stream_if_not_exists()
+ .create_topic_if_not_exists(
+ self.config.partitions,
+ Some(1),
+ IggyExpiry::NeverExpire,
+ MaxTopicSize::ServerDefault,
+ )
+ .build(),
+ );
+
+ self.producer.as_mut().unwrap().init().await?;
+ Ok(())
+ }
+}
+impl ApiLabel for HighLevelProducerClient {
+ const API_LABEL: &'static str = "high-level";
+}
+impl BenchmarkProducerClient for HighLevelProducerClient {}
diff --git a/core/bench/src/actors/producing_consumer/mod.rs b/core/bench/src/actors/producer/client/interface.rs
similarity index 51%
copy from core/bench/src/actors/producing_consumer/mod.rs
copy to core/bench/src/actors/producer/client/interface.rs
index 2927c8d9..f2dd6823 100644
--- a/core/bench/src/actors/producing_consumer/mod.rs
+++ b/core/bench/src/actors/producer/client/interface.rs
@@ -16,6 +16,28 @@
* under the License.
*/
-pub mod benchmark_producing_consumer;
+use crate::{
+ actors::{ApiLabel, BatchMetrics, BenchmarkInit},
+ utils::batch_generator::BenchmarkBatchGenerator,
+};
+use bench_report::numeric_parameter::BenchmarkNumericParameter;
+use iggy::prelude::*;
-pub use benchmark_producing_consumer::BenchmarkProducingConsumer;
+#[derive(Debug, Clone)]
+pub struct BenchmarkProducerConfig {
+ pub producer_id: u32,
+ pub stream_id: u32,
+ pub partitions: u32,
+ pub messages_per_batch: BenchmarkNumericParameter,
+ pub message_size: BenchmarkNumericParameter,
+ pub warmup_time: IggyDuration,
+}
+
+pub trait ProducerClient: Send + Sync {
+ async fn produce_batch(
+ &mut self,
+ batch_generator: &mut BenchmarkBatchGenerator,
+ ) -> Result<Option<BatchMetrics>, IggyError>;
+}
+
+pub trait BenchmarkProducerClient: ProducerClient + BenchmarkInit + ApiLabel + Send + Sync {}
diff --git a/core/bench/src/actors/producer/client/low_level.rs b/core/bench/src/actors/producer/client/low_level.rs
new file mode 100644
index 00000000..c657f354
--- /dev/null
+++ b/core/bench/src/actors/producer/client/low_level.rs
@@ -0,0 +1,113 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use std::sync::Arc;
+
+use crate::{
+ actors::{
+ ApiLabel, BatchMetrics, BenchmarkInit,
+ producer::client::{
+ BenchmarkProducerClient,
+ interface::{BenchmarkProducerConfig, ProducerClient},
+ },
+ },
+ utils::batch_generator::BenchmarkBatchGenerator,
+};
+use iggy::prelude::*;
+use integration::test_server::{ClientFactory, login_root};
+use tokio::time::Instant;
+
+pub struct LowLevelProducerClient {
+ client_factory: Arc<dyn ClientFactory>,
+ config: BenchmarkProducerConfig,
+ client: Option<IggyClient>,
+ stream_id: Identifier,
+ topic_id: Identifier,
+ partitioning: Partitioning,
+}
+
+impl LowLevelProducerClient {
+ pub fn new(client_factory: Arc<dyn ClientFactory>, config: BenchmarkProducerConfig) -> Self {
+ Self {
+ client_factory,
+ config,
+ client: None,
+ stream_id: Identifier::default(),
+ topic_id: Identifier::default(),
+ partitioning: Partitioning::partition_id(1),
+ }
+ }
+}
+
+impl ProducerClient for LowLevelProducerClient {
+ async fn produce_batch(
+ &mut self,
+ batch_generator: &mut BenchmarkBatchGenerator,
+ ) -> Result<Option<BatchMetrics>, IggyError> {
+ let client = self.client.as_mut().unwrap();
+ let batch = batch_generator.generate_batch();
+ if batch.messages.is_empty() {
+ return Ok(None);
+ }
+
+ let before_send = Instant::now();
+ client
+ .send_messages(
+ &self.stream_id,
+ &self.topic_id,
+ &self.partitioning,
+ &mut batch.messages,
+ )
+ .await?;
+ let latency = before_send.elapsed();
+
+ Ok(Some(BatchMetrics {
+ messages: u32::try_from(batch.messages.len()).unwrap(),
+ user_data_bytes: batch.user_data_bytes,
+ total_bytes: batch.total_bytes,
+ latency,
+ }))
+ }
+}
+
+impl BenchmarkInit for LowLevelProducerClient {
+ async fn setup(&mut self) -> Result<(), IggyError> {
+ let default_partition_id = 1;
+ let partitions = self.config.partitions;
+
+ let client = self.client_factory.create_client().await;
+ let client = IggyClient::create(client, None, None);
+ login_root(&client).await;
+
+ let partitioning = match partitions {
+ 0 => panic!("Partition count must be greater than 0"),
+ 1 => Partitioning::partition_id(default_partition_id),
+ _ => Partitioning::balanced(),
+ };
+
+ self.client = Some(client);
+ self.partitioning = partitioning;
+ self.stream_id = self.config.stream_id.try_into()?;
+ self.topic_id = Identifier::numeric(1)?;
+ Ok(())
+ }
+}
+impl ApiLabel for LowLevelProducerClient {
+ const API_LABEL: &'static str = "low-level";
+}
+impl BenchmarkProducerClient for LowLevelProducerClient {}
diff --git a/core/bench/src/actors/producer/mod.rs b/core/bench/src/actors/producer/client/mod.rs
similarity index 88%
copy from core/bench/src/actors/producer/mod.rs
copy to core/bench/src/actors/producer/client/mod.rs
index d4f6209f..20280351 100644
--- a/core/bench/src/actors/producer/mod.rs
+++ b/core/bench/src/actors/producer/client/mod.rs
@@ -16,6 +16,8 @@
* under the License.
*/
-pub mod benchmark_producer;
+pub mod high_level;
+pub mod interface;
+pub mod low_level;
-pub use benchmark_producer::BenchmarkProducer;
+pub use interface::BenchmarkProducerClient;
diff --git a/core/bench/src/actors/producer/mod.rs b/core/bench/src/actors/producer/mod.rs
index d4f6209f..36d70119 100644
--- a/core/bench/src/actors/producer/mod.rs
+++ b/core/bench/src/actors/producer/mod.rs
@@ -17,5 +17,6 @@
*/
pub mod benchmark_producer;
-
+pub mod client;
+pub mod typed_benchmark_producer;
pub use benchmark_producer::BenchmarkProducer;
diff --git a/core/bench/src/actors/producer/typed_benchmark_producer.rs b/core/bench/src/actors/producer/typed_benchmark_producer.rs
new file mode 100644
index 00000000..ee2de483
--- /dev/null
+++ b/core/bench/src/actors/producer/typed_benchmark_producer.rs
@@ -0,0 +1,99 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use crate::{
+ actors::producer::{
+ BenchmarkProducer,
+ client::{
+ high_level::HighLevelProducerClient, interface::BenchmarkProducerConfig,
+ low_level::LowLevelProducerClient,
+ },
+ },
+ utils::finish_condition::BenchmarkFinishCondition,
+};
+use bench_report::{
+ benchmark_kind::BenchmarkKind, individual_metrics::BenchmarkIndividualMetrics,
+ numeric_parameter::BenchmarkNumericParameter,
+};
+use iggy::prelude::*;
+use integration::test_server::ClientFactory;
+use std::sync::Arc;
+
+pub enum TypedBenchmarkProducer {
+ High(BenchmarkProducer<HighLevelProducerClient>),
+ Low(BenchmarkProducer<LowLevelProducerClient>),
+}
+
+impl TypedBenchmarkProducer {
+ #[allow(clippy::too_many_arguments)]
+ pub fn new(
+ use_high_level_api: bool,
+ client_factory: Arc<dyn ClientFactory>,
+ benchmark_kind: BenchmarkKind,
+ producer_id: u32,
+ stream_id: u32,
+ partitions: u32,
+ messages_per_batch: BenchmarkNumericParameter,
+ message_size: BenchmarkNumericParameter,
+ finish_condition: Arc<BenchmarkFinishCondition>,
+ warmup_time: IggyDuration,
+ sampling_time: IggyDuration,
+ moving_average_window: u32,
+ limit_bytes_per_second: Option<IggyByteSize>,
+ ) -> Self {
+ let config = BenchmarkProducerConfig {
+ producer_id,
+ stream_id,
+ partitions,
+ messages_per_batch,
+ message_size,
+ warmup_time,
+ };
+
+ if use_high_level_api {
+ let client = HighLevelProducerClient::new(client_factory, config.clone());
+ Self::High(BenchmarkProducer::new(
+ client,
+ benchmark_kind,
+ finish_condition,
+ sampling_time,
+ moving_average_window,
+ limit_bytes_per_second,
+ config,
+ ))
+ } else {
+ let client = LowLevelProducerClient::new(client_factory, config.clone());
+ Self::Low(BenchmarkProducer::new(
+ client,
+ benchmark_kind,
+ finish_condition,
+ sampling_time,
+ moving_average_window,
+ limit_bytes_per_second,
+ config,
+ ))
+ }
+ }
+
+ pub async fn run(self) -> Result<BenchmarkIndividualMetrics, IggyError> {
+ match self {
+ Self::High(producer) => producer.run().await,
+ Self::Low(producer) => producer.run().await,
+ }
+ }
+}
diff --git a/core/bench/src/actors/producing_consumer/benchmark_producing_consumer.rs b/core/bench/src/actors/producing_consumer/benchmark_producing_consumer.rs
index 2969c4a6..0f5d323a 100644
--- a/core/bench/src/actors/producing_consumer/benchmark_producing_consumer.rs
+++ b/core/bench/src/actors/producing_consumer/benchmark_producing_consumer.rs
@@ -16,437 +16,193 @@
* under the License.
*/
-use crate::analytics::record::BenchmarkRecord;
-use crate::benchmarks::common::create_consumer;
-use crate::utils::batch_generator::BenchmarkBatchGenerator;
-use crate::utils::finish_condition::BenchmarkFinishCondition;
-use crate::utils::{batch_total_size_bytes, batch_user_size_bytes};
+use std::{sync::Arc, time::Duration};
+
use crate::{
- analytics::metrics::individual::from_records, utils::rate_limiter::BenchmarkRateLimiter,
+ actors::{
+ consumer::client::{BenchmarkConsumerClient, interface::BenchmarkConsumerConfig},
+ producer::client::{BenchmarkProducerClient, interface::BenchmarkProducerConfig},
+ },
+ analytics::{metrics::individual::from_records, record::BenchmarkRecord},
+ utils::{
+ batch_generator::BenchmarkBatchGenerator, finish_condition::BenchmarkFinishCondition,
+ rate_limiter::BenchmarkRateLimiter,
+ },
+};
+use bench_report::{
+ actor_kind::ActorKind, benchmark_kind::BenchmarkKind,
+ individual_metrics::BenchmarkIndividualMetrics, numeric_parameter::BenchmarkNumericParameter,
};
-use bench_report::actor_kind::ActorKind;
-use bench_report::benchmark_kind::BenchmarkKind;
-use bench_report::individual_metrics::BenchmarkIndividualMetrics;
-use bench_report::numeric_parameter::BenchmarkNumericParameter;
use human_repr::HumanCount;
use iggy::prelude::*;
-use integration::test_server::{ClientFactory, login_root};
-use std::sync::Arc;
-use std::time::Duration;
use tokio::time::Instant;
-use tracing::{info, warn};
-
-pub struct BenchmarkProducingConsumer {
- client_factory: Arc<dyn ClientFactory>,
- benchmark_kind: BenchmarkKind,
- actor_id: u32,
- consumer_group_id: Option<u32>,
- stream_id: u32,
- partitions_count: u32,
- messages_per_batch: BenchmarkNumericParameter,
- message_size: BenchmarkNumericParameter,
- send_finish_condition: Arc<BenchmarkFinishCondition>,
- poll_finish_condition: Arc<BenchmarkFinishCondition>,
- warmup_time: IggyDuration,
- sampling_time: IggyDuration,
- moving_average_window: u32,
- limit_bytes_per_second: Option<IggyByteSize>,
- polling_kind: PollingKind,
+use tracing::info;
+
+pub struct BenchmarkProducingConsumer<P, C>
+where
+ P: BenchmarkProducerClient,
+ C: BenchmarkConsumerClient,
+{
+ pub producer: P,
+ pub consumer: C,
+ pub benchmark_kind: BenchmarkKind,
+ pub send_finish_condition: Arc<BenchmarkFinishCondition>,
+ pub poll_finish_condition: Arc<BenchmarkFinishCondition>,
+ pub sampling_time: IggyDuration,
+ pub moving_average_window: u32,
+ pub limit_bytes_per_second: Option<IggyByteSize>,
+ pub producer_config: BenchmarkProducerConfig,
+ pub consumer_config: BenchmarkConsumerConfig,
}
-impl BenchmarkProducingConsumer {
+impl<P, C> BenchmarkProducingConsumer<P, C>
+where
+ P: BenchmarkProducerClient,
+ C: BenchmarkConsumerClient,
+{
#[allow(clippy::too_many_arguments)]
- pub fn new(
- client_factory: Arc<dyn ClientFactory>,
+ pub const fn new(
+ producer: P,
+ consumer: C,
benchmark_kind: BenchmarkKind,
- actor_id: u32,
- consumer_group_id: Option<u32>,
- stream_id: u32,
- partitions_count: u32,
- messages_per_batch: BenchmarkNumericParameter,
- message_size: BenchmarkNumericParameter,
send_finish_condition: Arc<BenchmarkFinishCondition>,
poll_finish_condition: Arc<BenchmarkFinishCondition>,
- warmup_time: IggyDuration,
sampling_time: IggyDuration,
moving_average_window: u32,
limit_bytes_per_second: Option<IggyByteSize>,
- polling_kind: PollingKind,
+ producer_config: BenchmarkProducerConfig,
+ consumer_config: BenchmarkConsumerConfig,
) -> Self {
Self {
- client_factory,
+ producer,
+ consumer,
benchmark_kind,
- actor_id,
- consumer_group_id,
- stream_id,
- partitions_count,
- messages_per_batch,
- message_size,
send_finish_condition,
poll_finish_condition,
- warmup_time,
sampling_time,
moving_average_window,
limit_bytes_per_second,
- polling_kind,
+ producer_config,
+ consumer_config,
}
}
-
- pub async fn run(self) -> Result<BenchmarkIndividualMetrics, IggyError> {
- let (
- client,
- stream_id,
- topic_id,
- partitioning,
- partition_id,
- consumer,
- mut batch_generator,
- rate_limiter,
- ) = self.setup_client().await?;
-
- if self.warmup_time.get_duration() != Duration::from_millis(0) {
- self.run_warmup(
- &client,
- &stream_id,
- &topic_id,
- &partitioning,
- partition_id,
- &consumer,
- &mut batch_generator,
- )
- .await?;
- }
-
- let metrics = self
- .run_benchmark(
- client,
- stream_id,
- topic_id,
- partitioning,
- partition_id,
- consumer,
- batch_generator,
- rate_limiter,
- )
- .await?;
- Ok(metrics)
- }
-
- async fn setup_client(
- &self,
- ) -> Result<
- (
- IggyClient,
- Identifier,
- Identifier,
- Partitioning,
- Option<u32>,
- Consumer,
- BenchmarkBatchGenerator,
- Option<BenchmarkRateLimiter>,
- ),
- IggyError,
- > {
- let topic_id: u32 = 1;
- let default_partition_id: u32 = 1;
-
- let client = self.client_factory.create_client().await;
- let client = IggyClient::create(client, None, None);
- login_root(&client).await;
-
- let stream_id = self.stream_id.try_into()?;
- let topic_id = topic_id.try_into()?;
- let partitioning = match self.partitions_count {
- 0 => panic!("Partition count must be greater than 0"),
- 1 => Partitioning::partition_id(default_partition_id),
- 2.. => Partitioning::balanced(),
- };
- let partition_id = if self.consumer_group_id.is_some() {
- None
- } else {
- Some(default_partition_id)
- };
-
- let consumer = create_consumer(
- &client,
- self.consumer_group_id.as_ref(),
- &stream_id,
- &topic_id,
- self.actor_id,
- )
- .await;
-
- let batch_generator =
- BenchmarkBatchGenerator::new(self.message_size, self.messages_per_batch);
+ #[allow(clippy::too_many_lines)]
+ #[allow(clippy::cognitive_complexity)]
+ pub async fn run(mut self) -> Result<BenchmarkIndividualMetrics, IggyError> {
+ info!(
+ "ProducingConsumer #{producer_id} → sending {send_status} and polling {poll_status} ({messages_per_batch} msgs/batch) on stream {stream_id}, rate limit: {rate_limit:?}",
+ producer_id = self.producer_config.producer_id,
+ send_status = self.send_finish_condition.total_str(),
+ poll_status = self.poll_finish_condition.total_str(),
+ messages_per_batch = self.producer_config.messages_per_batch,
+ stream_id = self.producer_config.stream_id,
+ rate_limit = self.limit_bytes_per_second,
+ );
+ self.producer.setup().await?;
+ self.consumer.setup().await?;
+ let mut batch_generator = BenchmarkBatchGenerator::new(
+ self.producer_config.message_size,
+ self.producer_config.messages_per_batch,
+ );
let rate_limiter = self.limit_bytes_per_second.map(BenchmarkRateLimiter::new);
+ if self.producer_config.warmup_time.get_duration() != Duration::from_millis(0) {
+ self.log_warmup_info();
+ let warmup_end = Instant::now() + self.producer_config.warmup_time.get_duration();
- Ok((
- client,
- stream_id,
- topic_id,
- partitioning,
- partition_id,
- consumer,
- batch_generator,
- rate_limiter,
- ))
- }
-
- #[allow(clippy::too_many_arguments, clippy::cognitive_complexity)]
- async fn run_warmup(
- &self,
- client: &IggyClient,
- stream_id: &Identifier,
- topic_id: &Identifier,
- partitioning: &Partitioning,
- partition_id: Option<u32>,
- consumer: &Consumer,
- batch_generator: &mut BenchmarkBatchGenerator,
- ) -> Result<(), IggyError> {
- let warmup_end = Instant::now() + self.warmup_time.get_duration();
- let mut offset: u64 = 0;
- let mut last_warning_time: Option<Instant> = None;
- let mut skipped_warnings_count: u32 = 0;
-
- if let Some(cg_id) = self.consumer_group_id {
- info!(
- "ProducingConsumer #{}, part of consumer group #{}, → warming up for {}...",
- self.actor_id, cg_id, self.warmup_time
- );
- } else {
- info!(
- "ProducingConsumer #{} → warming up for {}...",
- self.actor_id, self.warmup_time
- );
- }
- while Instant::now() < warmup_end {
- let batch = batch_generator.generate_batch();
- client
- .send_messages(stream_id, topic_id, partitioning, &mut batch.messages)
- .await?;
-
- let (strategy, auto_commit) = match self.polling_kind {
- PollingKind::Offset => (PollingStrategy::offset(offset), false),
- PollingKind::Next => (PollingStrategy::next(), true),
- other => panic!("Unsupported polling kind for warmup: {other:?}"),
- };
-
- let polled_messages = client
- .poll_messages(
- stream_id,
- topic_id,
- partition_id,
- consumer,
- &strategy,
- u32::try_from(batch.messages.len()).unwrap_or(u32::MAX),
- auto_commit,
- )
- .await?;
-
- if polled_messages.messages.is_empty() {
- let should_warn =
- last_warning_time.is_none_or(|t| t.elapsed() >= Duration::from_secs(1));
-
- if should_warn {
- warn!(
- "ProducingConsumer #{actor_id} (warmup) → expected {messages_per_batch} messages but got {actual_count}, retrying... ({skipped_warnings_count} warnings skipped)",
- actor_id = self.actor_id,
- messages_per_batch = self.messages_per_batch,
- actual_count = polled_messages.messages.len()
- );
- last_warning_time = Some(Instant::now());
- skipped_warnings_count = 0;
- } else {
- skipped_warnings_count += 1;
- }
- continue;
+ while Instant::now() < warmup_end {
+ let _ = self.producer.produce_batch(&mut batch_generator).await;
+ let _ = self.consumer.consume_batch().await;
}
-
- offset += u64::try_from(batch.messages.len()).unwrap_or(u64::MAX);
}
- Ok(())
- }
-
- #[allow(
- clippy::too_many_arguments,
- clippy::too_many_lines,
- clippy::cognitive_complexity
- )]
- async fn run_benchmark(
- self,
- client: IggyClient,
- stream_id: Identifier,
- topic_id: Identifier,
- partitioning: Partitioning,
- partition_id: Option<u32>,
- consumer: Consumer,
- mut batch_generator: BenchmarkBatchGenerator,
- rate_limiter: Option<BenchmarkRateLimiter>,
- ) -> Result<BenchmarkIndividualMetrics, IggyError> {
- info!(
- "ProducingConsumer #{} → sending {} and polling {} ({} msgs/batch) on stream {}, rate limit: {:?}",
- self.actor_id,
- self.send_finish_condition.total_str(),
- self.poll_finish_condition.total_str(),
- self.messages_per_batch,
- stream_id,
- self.limit_bytes_per_second
- );
+ self.log_setup_info();
let max_capacity = self
.send_finish_condition
.max_capacity()
.max(self.poll_finish_condition.max_capacity());
- let mut records: Vec<BenchmarkRecord> = Vec::with_capacity(max_capacity);
- let mut offset = 0;
- let mut last_warning_time: Option<Instant> = None;
- let mut skipped_warnings_count = 0;
-
- let mut total_user_data_bytes_processed;
- let mut total_bytes_processed;
- let mut total_messages_processed = 0;
- let mut total_batches_processed = 0;
+ let mut records = Vec::with_capacity(max_capacity);
- let mut sent_user_bytes_processed = 0;
- let mut sent_total_bytes_processed = 0;
+ let mut rl_value = 0;
+ let mut sent_user_bytes = 0;
+ let mut sent_total_bytes = 0;
let mut sent_messages = 0;
let mut sent_batches = 0;
- let mut received_user_bytes_processed = 0;
- let mut received_total_bytes_processed = 0;
- let mut received_messages = 0;
- let mut received_batches = 0;
-
- let mut last_received_batch_user_data_bytes;
- let mut rl_value = 0;
+ let mut recv_user_bytes = 0;
+ let mut recv_total_bytes = 0;
+ let mut recv_messages = 0;
+ let mut recv_batches = 0;
let is_producer = self.send_finish_condition.total() > 0;
let is_consumer = self.poll_finish_condition.total() > 0;
- let require_reply = is_producer && is_consumer && self.consumer_group_id.is_none();
+ let require_reply =
+ is_producer && is_consumer && self.consumer_config.consumer_group_id.is_none();
let mut awaiting_reply = false;
- let start_timestamp = Instant::now();
+ let start = Instant::now();
while !(self.send_finish_condition.is_done() && self.poll_finish_condition.is_done()) {
- let may_send = is_producer
+ if is_producer
&& !self.send_finish_condition.is_done()
- && (!require_reply || !awaiting_reply);
-
- if may_send {
- let batch = batch_generator.generate_batch();
- client
- .send_messages(&stream_id, &topic_id, &partitioning, &mut batch.messages)
- .await?;
-
- rl_value += batch.user_data_bytes;
-
- sent_total_bytes_processed += batch.total_bytes;
- sent_user_bytes_processed += batch.user_data_bytes;
- sent_messages += u32::try_from(batch.messages.len()).unwrap_or(u32::MAX);
- sent_batches += 1;
-
- awaiting_reply = is_consumer;
-
- if self
- .send_finish_condition
- .account_and_check(batch.user_data_bytes)
- {
- info!(
- "ProducingConsumer #{actor_id} → finished sending {sent_messages} messages in {sent_batches} batches ({sent_user_bytes_processed} bytes of user data, {sent_total_bytes_processed} bytes of total data), send finish condition: {send_status}, poll finish condition: {poll_status}",
- actor_id = self.actor_id,
- sent_messages = sent_messages.human_count_bare(),
- sent_batches = sent_batches.human_count_bare(),
- sent_user_bytes_processed = sent_user_bytes_processed.human_count_bytes(),
- sent_total_bytes_processed = sent_total_bytes_processed.human_count_bytes(),
- send_status = self.send_finish_condition.status(),
- poll_status = self.poll_finish_condition.status()
- );
- }
- }
-
- if is_consumer && !self.poll_finish_condition.is_done() {
- let (strategy, auto_commit) = match self.polling_kind {
- PollingKind::Offset => (PollingStrategy::offset(offset), false),
- PollingKind::Next => (PollingStrategy::next(), true),
- other => panic!("Unsupported polling kind for benchmark: {other:?}"),
- };
-
- let polled_messages = client
- .poll_messages(
- &stream_id,
- &topic_id,
- partition_id,
- &consumer,
- &strategy,
- self.messages_per_batch.max(),
- auto_commit,
- )
- .await?;
-
- if polled_messages.messages.is_empty() {
- let should_warn =
- last_warning_time.is_none_or(|t| t.elapsed() >= Duration::from_secs(1));
-
- if should_warn {
- warn!(
- "ProducingConsumer #{actor_id} → received empty batch, sent: {send_status}, polled: {poll_status}, polling kind: {polling_kind:?}, retrying... ({skipped_warnings_count} warnings skipped in last second)",
- actor_id = self.actor_id,
+ && (!require_reply || !awaiting_reply)
+ {
+ if let Some(batch) = self.producer.produce_batch(&mut batch_generator).await? {
+ rl_value += batch.user_data_bytes;
+ sent_user_bytes += batch.user_data_bytes;
+ sent_total_bytes += batch.total_bytes;
+ sent_messages += u64::from(batch.messages);
+ sent_batches += 1;
+ awaiting_reply = is_consumer;
+
+ if self
+ .send_finish_condition
+ .account_and_check(batch.user_data_bytes)
+ {
+ info!(
+ "ProducingConsumer #{actor_id} → finished sending {sent_messages} messages in {sent_batches} batches ({sent_user_bytes} bytes of user data, {sent_total_bytes} bytes of total data), send finish condition: {send_status}, poll finish condition: {poll_status}",
+ actor_id = self.producer_config.producer_id,
+ sent_messages = sent_messages.human_count_bare(),
+ sent_batches = sent_batches.human_count_bare(),
+ sent_user_bytes = sent_user_bytes.human_count_bytes(),
+ sent_total_bytes = sent_total_bytes.human_count_bytes(),
send_status = self.send_finish_condition.status(),
- poll_status = self.poll_finish_condition.status(),
- polling_kind = self.polling_kind
+ poll_status = self.poll_finish_condition.status()
);
- last_warning_time = Some(Instant::now());
- skipped_warnings_count = 0;
- } else {
- skipped_warnings_count += 1;
}
-
- continue;
- }
-
- let now = IggyTimestamp::now().as_micros();
- let latency = Duration::from_micros(
- now - polled_messages.messages[0].header.origin_timestamp,
- );
-
- last_received_batch_user_data_bytes = batch_user_size_bytes(&polled_messages);
- rl_value += last_received_batch_user_data_bytes;
-
- received_user_bytes_processed += last_received_batch_user_data_bytes;
- received_total_bytes_processed += batch_total_size_bytes(&polled_messages);
- received_messages +=
- u32::try_from(polled_messages.messages.len()).unwrap_or(u32::MAX);
- received_batches += 1;
-
- offset += u64::try_from(polled_messages.messages.len()).unwrap_or(u64::MAX);
-
- total_user_data_bytes_processed =
- received_user_bytes_processed + sent_user_bytes_processed;
- total_bytes_processed = received_total_bytes_processed + sent_total_bytes_processed;
- total_messages_processed = received_messages + sent_messages;
- total_batches_processed = received_batches + sent_batches;
-
- records.push(BenchmarkRecord {
- elapsed_time_us: u64::try_from(start_timestamp.elapsed().as_micros())
- .unwrap_or(u64::MAX),
- latency_us: u64::try_from(latency.as_micros()).unwrap_or(u64::MAX),
- messages: u64::from(total_messages_processed),
- message_batches: total_batches_processed,
- user_data_bytes: total_user_data_bytes_processed,
- total_bytes: total_bytes_processed,
- });
-
- if let Some(rate_limiter) = &rate_limiter {
- rate_limiter.wait_until_necessary(rl_value).await;
- rl_value = 0;
}
+ }
- self.poll_finish_condition
- .account_and_check(last_received_batch_user_data_bytes);
+ if is_consumer && !self.poll_finish_condition.is_done() {
+ if let Some(batch) = self.consumer.consume_batch().await? {
+ rl_value += batch.user_data_bytes;
+ recv_user_bytes += batch.user_data_bytes;
+ recv_total_bytes += batch.total_bytes;
+ recv_messages += u64::from(batch.messages);
+ recv_batches += 1;
+
+ let elapsed = u64::try_from(start.elapsed().as_micros()).unwrap_or(u64::MAX);
+ let latency = u64::try_from(batch.latency.as_micros()).unwrap_or(u64::MAX);
+
+ records.push(BenchmarkRecord {
+ elapsed_time_us: elapsed,
+ latency_us: latency,
+ messages: sent_messages + recv_messages,
+ message_batches: sent_batches + recv_batches,
+ user_data_bytes: sent_user_bytes + recv_user_bytes,
+ total_bytes: sent_total_bytes + recv_total_bytes,
+ });
+
+ if let Some(limiter) = &rate_limiter {
+ limiter.wait_until_necessary(rl_value).await;
+ rl_value = 0;
+ }
- if require_reply {
- awaiting_reply = false;
+ self.poll_finish_condition
+ .account_and_check(batch.user_data_bytes);
+ if require_reply {
+ awaiting_reply = false;
+ }
}
}
}
@@ -455,21 +211,60 @@ impl BenchmarkProducingConsumer {
&records,
self.benchmark_kind,
ActorKind::ProducingConsumer,
- self.actor_id,
+ self.producer_config.producer_id,
self.sampling_time,
self.moving_average_window,
);
Self::log_statistics(
- self.actor_id,
- u64::from(total_messages_processed),
- total_batches_processed,
- &self.messages_per_batch,
+ self.producer_config.producer_id,
+ recv_messages,
+ recv_batches,
+ &self.producer_config.messages_per_batch,
&metrics,
);
Ok(metrics)
}
+ fn log_setup_info(&self) {
+ if let Some(cg_id) = self.consumer_config.consumer_group_id {
+ info!(
+ "ProducingConsumer #{} → sending & polling {} messages per batch ({} bytes each) on stream {} with {} partition(s), part of CG #{} ({})",
+ self.producer_config.producer_id,
+ self.producer_config.messages_per_batch,
+ self.producer_config.message_size,
+ self.producer_config.stream_id,
+ self.producer_config.partitions,
+ cg_id,
+ C::API_LABEL
+ );
+ } else {
+ info!(
+ "ProducingConsumer #{} → sending & polling {} messages per batch ({} bytes each) on stream {} with {} partition(s) ({})",
+ self.producer_config.producer_id,
+ self.producer_config.messages_per_batch,
+ self.producer_config.message_size,
+ self.producer_config.stream_id,
+ self.producer_config.partitions,
+ C::API_LABEL
+ );
+ }
+ }
+
+ fn log_warmup_info(&self) {
+ if let Some(cg_id) = self.consumer_config.consumer_group_id {
+ info!(
+ "ProducingConsumer #{}, part of consumer group #{} → warming up for {}...",
+ self.consumer_config.consumer_id, cg_id, self.consumer_config.warmup_time
+ );
+ } else {
+ info!(
+ "ProducingConsumer #{} → warming up for {}...",
+ self.consumer_config.consumer_id, self.consumer_config.warmup_time
+ );
+ }
+ }
+
fn log_statistics(
actor_id: u32,
total_messages: u64,
diff --git a/core/bench/src/actors/producing_consumer/mod.rs b/core/bench/src/actors/producing_consumer/mod.rs
index 2927c8d9..165a6f88 100644
--- a/core/bench/src/actors/producing_consumer/mod.rs
+++ b/core/bench/src/actors/producing_consumer/mod.rs
@@ -17,5 +17,5 @@
*/
pub mod benchmark_producing_consumer;
-
+pub mod typed_banchmark_producing_consumer;
pub use benchmark_producing_consumer::BenchmarkProducingConsumer;
diff --git a/core/bench/src/actors/producing_consumer/typed_banchmark_producing_consumer.rs b/core/bench/src/actors/producing_consumer/typed_banchmark_producing_consumer.rs
new file mode 100644
index 00000000..26e3e0fb
--- /dev/null
+++ b/core/bench/src/actors/producing_consumer/typed_banchmark_producing_consumer.rs
@@ -0,0 +1,128 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use std::sync::Arc;
+
+use crate::{
+ actors::{
+ consumer::client::{
+ high_level::HighLevelConsumerClient, interface::BenchmarkConsumerConfig,
+ low_level::LowLevelConsumerClient,
+ },
+ producer::client::{
+ high_level::HighLevelProducerClient, interface::BenchmarkProducerConfig,
+ low_level::LowLevelProducerClient,
+ },
+ producing_consumer::BenchmarkProducingConsumer,
+ },
+ utils::finish_condition::BenchmarkFinishCondition,
+};
+use bench_report::{
+ benchmark_kind::BenchmarkKind, individual_metrics::BenchmarkIndividualMetrics,
+ numeric_parameter::BenchmarkNumericParameter,
+};
+
+use iggy::prelude::*;
+use integration::test_server::ClientFactory;
+
+pub enum TypedBenchmarkProducingConsumer {
+ High(BenchmarkProducingConsumer<HighLevelProducerClient, HighLevelConsumerClient>),
+ Low(BenchmarkProducingConsumer<LowLevelProducerClient, LowLevelConsumerClient>),
+}
+impl TypedBenchmarkProducingConsumer {
+ #[allow(clippy::too_many_arguments)]
+ pub fn new(
+ use_high_level_api: bool,
+ client_factory: Arc<dyn ClientFactory>,
+ benchmark_kind: BenchmarkKind,
+ actor_id: u32,
+ consumer_group_id: Option<u32>,
+ stream_id: u32,
+ partitions_count: u32,
+ messages_per_batch: BenchmarkNumericParameter,
+ message_size: BenchmarkNumericParameter,
+ send_finish_condition: Arc<BenchmarkFinishCondition>,
+ poll_finish_condition: Arc<BenchmarkFinishCondition>,
+ warmup_time: IggyDuration,
+ sampling_time: IggyDuration,
+ moving_average_window: u32,
+ limit_bytes_per_second: Option<IggyByteSize>,
+ polling_kind: PollingKind,
+ origin_timestamp_latency_calculation: bool,
+ ) -> Self {
+ let producer_config = BenchmarkProducerConfig {
+ producer_id: actor_id,
+ stream_id,
+ partitions: partitions_count,
+ messages_per_batch,
+ message_size,
+ warmup_time,
+ };
+
+ let consumer_config = BenchmarkConsumerConfig {
+ consumer_id: actor_id,
+ consumer_group_id,
+ stream_id,
+ messages_per_batch,
+ warmup_time,
+ polling_kind,
+ origin_timestamp_latency_calculation,
+ };
+
+ if use_high_level_api {
+ let producer =
+ HighLevelProducerClient::new(client_factory.clone(), producer_config.clone());
+ let consumer = HighLevelConsumerClient::new(client_factory, consumer_config.clone());
+ Self::High(BenchmarkProducingConsumer::new(
+ producer,
+ consumer,
+ benchmark_kind,
+ send_finish_condition,
+ poll_finish_condition,
+ sampling_time,
+ moving_average_window,
+ limit_bytes_per_second,
+ producer_config,
+ consumer_config,
+ ))
+ } else {
+ let producer =
+ LowLevelProducerClient::new(client_factory.clone(), producer_config.clone());
+ let consumer = LowLevelConsumerClient::new(client_factory, consumer_config.clone());
+ Self::Low(BenchmarkProducingConsumer::new(
+ producer,
+ consumer,
+ benchmark_kind,
+ send_finish_condition,
+ poll_finish_condition,
+ sampling_time,
+ moving_average_window,
+ limit_bytes_per_second,
+ producer_config,
+ consumer_config,
+ ))
+ }
+ }
+
+ pub async fn run(self) -> Result<BenchmarkIndividualMetrics, IggyError> {
+ match self {
+ Self::High(p) => p.run().await,
+ Self::Low(p) => p.run().await,
+ }
+ }
+}
diff --git a/core/bench/src/benchmarks/common.rs b/core/bench/src/benchmarks/common.rs
index 6c774443..d5436b56 100644
--- a/core/bench/src/benchmarks/common.rs
+++ b/core/bench/src/benchmarks/common.rs
@@ -19,8 +19,9 @@
use super::{CONSUMER_GROUP_BASE_ID, CONSUMER_GROUP_NAME_PREFIX};
use crate::{
actors::{
- consumer::BenchmarkConsumer, producer::BenchmarkProducer,
- producing_consumer::BenchmarkProducingConsumer,
+ consumer::typed_benchmark_consumer::TypedBenchmarkConsumer,
+ producer::typed_benchmark_producer::TypedBenchmarkProducer,
+ producing_consumer::typed_banchmark_producing_consumer::TypedBenchmarkProducingConsumer,
},
args::common::IggyBenchArgs,
utils::finish_condition::{BenchmarkFinishCondition, BenchmarkFinishConditionMode},
@@ -128,6 +129,7 @@ pub fn build_producer_futures(
let shared_finish_condition =
BenchmarkFinishCondition::new(args, BenchmarkFinishConditionMode::Shared);
let rate_limit = rate_limit_per_actor(args.rate_limit(), actors);
+ let use_high_level_api = args.high_level_api();
(1..=producers)
.map(|producer_id| {
@@ -142,7 +144,8 @@ pub fn build_producer_futures(
let stream_id = start_stream_id + 1 + (producer_id % streams);
async move {
- let producer = BenchmarkProducer::new(
+ let producer = TypedBenchmarkProducer::new(
+ use_high_level_api,
client_factory,
kind,
producer_id,
@@ -213,7 +216,8 @@ pub fn build_consumer_futures(
};
async move {
- let consumer = BenchmarkConsumer::new(
+ let consumer = TypedBenchmarkConsumer::new(
+ use_high_level_api,
client_factory,
kind,
consumer_id,
@@ -227,7 +231,6 @@ pub fn build_consumer_futures(
polling_kind,
rate_limit,
origin_timestamp_latency_calculation,
- use_high_level_api,
);
consumer.run().await
}
@@ -263,7 +266,15 @@ pub fn build_producing_consumers_futures(
&args,
BenchmarkFinishConditionMode::PerProducingConsumer,
);
-
+ let origin_timestamp_latency_calculation = match args.kind() {
+ BenchmarkKind::PinnedConsumer
+ | BenchmarkKind::PinnedProducerAndConsumer
+ | BenchmarkKind::BalancedConsumerGroup // TODO(hubcio): in future, PinnedProducerAndConsumer can also be true
+ | BenchmarkKind::EndToEndProducingConsumer => false,
+ BenchmarkKind::BalancedProducerAndConsumerGroup => true,
+ _ => unreachable!(),
+ };
+ let use_high_level_api = args.high_level_api();
let rate_limit = rate_limit_per_actor(args.rate_limit(), producing_consumers);
async move {
@@ -271,7 +282,8 @@ pub fn build_producing_consumers_futures(
"Executing producing consumer #{}, stream_id={}",
actor_id, stream_id
);
- let actor = BenchmarkProducingConsumer::new(
+ let actor = TypedBenchmarkProducingConsumer::new(
+ use_high_level_api,
client_factory_clone,
args_clone.kind(),
actor_id,
@@ -287,6 +299,7 @@ pub fn build_producing_consumers_futures(
args_clone.moving_average_window(),
rate_limit,
polling_kind,
+ origin_timestamp_latency_calculation,
);
actor.run().await
}
@@ -310,7 +323,7 @@ pub fn build_producing_consumer_groups_futures(
let start_stream_id = args.start_stream_id();
let start_consumer_group_id = CONSUMER_GROUP_BASE_ID;
let polling_kind = PollingKind::Next;
-
+ let use_high_level_api = args.high_level_api();
let shared_send_finish_condition =
BenchmarkFinishCondition::new(&args, BenchmarkFinishConditionMode::SharedHalf);
@@ -349,6 +362,14 @@ pub fn build_producing_consumer_groups_futures(
} else {
None
};
+ let origin_timestamp_latency_calculation = match args.kind() {
+ BenchmarkKind::PinnedConsumer
+ | BenchmarkKind::PinnedProducerAndConsumer
+ | BenchmarkKind::BalancedConsumerGroup // TODO(hubcio): in future, PinnedProducerAndConsumer can also be true
+ | BenchmarkKind::EndToEndProducingConsumer => false,
+ BenchmarkKind::BalancedProducerAndConsumerGroup => true,
+ _ => unreachable!(),
+ };
async move {
let actor_type = match (should_produce, should_consume) {
@@ -369,7 +390,8 @@ pub fn build_producing_consumer_groups_futures(
stream_id,
actor_type
);
- let actor = BenchmarkProducingConsumer::new(
+ let actor = TypedBenchmarkProducingConsumer::new(
+ use_high_level_api,
client_factory_clone,
args_clone.kind(),
actor_id,
@@ -385,7 +407,9 @@ pub fn build_producing_consumer_groups_futures(
args_clone.moving_average_window(),
rate_limit,
polling_kind,
+ origin_timestamp_latency_calculation,
);
+
actor.run().await
}
})
diff --git a/core/bench/src/utils/batch_generator.rs b/core/bench/src/utils/batch_generator.rs
index 22994bf1..e406a70c 100644
--- a/core/bench/src/utils/batch_generator.rs
+++ b/core/bench/src/utils/batch_generator.rs
@@ -113,6 +113,10 @@ impl BenchmarkBatchGenerator {
&mut self.batch
}
+ pub fn generate_owned_batch(&self) -> BenchmarkMessagesBatch {
+ self.build_single_fixed()
+ }
+
#[inline]
fn build_single_fixed(&self) -> BenchmarkMessagesBatch {
let payload_length = self.message_size.max();