feat: add fluvio producer callback #4340

fraidev · 2025-01-19T22:09:04Z

This PR adds a callback option for the producer.

Also prints total bytes and total records per sec at the end:

> fluvio bench producer --batch-size 16kib --num-records 300000
27244 records sent, 27244 records/sec: (138.9 KB/sec), 1.34ms avg latency, 3.69ms max latency
37800 records sent, 37800 records/sec: (192.8 KB/sec), 1.41ms avg latency, 10.63ms max latency
42028 records sent, 42028 records/sec: (214.3 KB/sec), 1.38ms avg latency, 10.63ms max latency
41552 records sent, 41552 records/sec: (211.9 KB/sec), 1.36ms avg latency, 10.63ms max latency
40138 records sent, 40138 records/sec: (204.7 KB/sec), 1.37ms avg latency, 12.28ms max latency
40868 records sent, 40868 records/sec: (208.4 KB/sec), 1.37ms avg latency, 12.28ms max latency
40488 records sent, 40488 records/sec: (206.5 KB/sec), 1.37ms avg latency, 12.28ms max latency

1.36ms avg latency, 12.28ms max latency, 1.27ms p0.50, 1.70ms p0.95, 2.54ms p0.99
300000 total records sent, 40689 records/sec: (207.5 MB/sec) 
Benchmark completed

sehz · 2025-01-20T05:16:25Z

crates/fluvio-benchmark/src/producer_worker.rs

-            self.stat.start();
-            let time = std::time::Instant::now();
-            let send_out = self
+            self.stat.start().await;


should remove this

Start is setting the first instant to calculate the final bandwidth at the end.

sehz · 2025-01-20T05:16:42Z

crates/fluvio-benchmark/src/producer_worker.rs

                .fluvio_producer
                .send(record.key, record.data.clone())
                .await?;

-            self.stat.send_out((send_out, time));
            self.stat.add_record(record.data.len() as u64).await;


no longer need this

sehz · 2025-01-20T05:18:44Z

crates/fluvio-benchmark/src/stats_collector.rs

-        }
+            // send end
+            let hist = hist.lock().await;
+            let start_time = start_time.lock().await.expect("start time");


start time should be tracked by callback?

This is the first start time, I can rename it

Using first start time from callback now

sehz · 2025-01-20T05:19:11Z

crates/fluvio-benchmark/src/stats_collector.rs

+            // send end
+            let hist = hist.lock().await;
+            let start_time = start_time.lock().await.expect("start time");
+            let elapsed = start_time.elapsed();


this should be come from callback

sehz · 2025-01-20T05:21:16Z

crates/fluvio/src/producer/config.rs

@@ -118,6 +120,9 @@ pub struct TopicProducerConfig {

    #[builder(default)]
    pub(crate) smartmodules: Vec<SmartModuleInvocation>,
+
+    #[builder(setter(into, strip_option), default)]
+    pub(crate) callback: Option<SharedProducerCallback<ProduceCompletionEvent>>,


why make generic? don't think there is need to make generic

Good catch!
It was to make it easy to handle lifetimes in my first implementation. But it's not needed for actual implementation.

I removed them

sehz · 2025-01-20T05:22:46Z

crates/fluvio/src/producer/partition_producer.rs

+                    created_at: batch_metadata.created_at,
+                    metadata: batch_metadata.clone(),
+                };
+                callback.finished(event).await.unwrap();


remove unwrap

sehz · 2025-01-20T05:25:49Z

crates/fluvio-benchmark/src/stats_collector.rs

    }

-    pub(crate) fn set_current_time(&mut self) {
-        self.start_time = Instant::now();
+    pub(crate) fn add_record(&mut self, bytes: u64) {


this should be invoked by callback

invoked by callback now

crates/fluvio/src/producer/accumulator.rs

sehz · 2025-01-20T19:01:06Z

crates/fluvio/src/producer/accumulator.rs

@@ -318,7 +332,8 @@ type ProduceResponseFuture = Shared<BoxFuture<'static, Arc<Result<ProduceRespons

 /// A Future that resolves to pair `base_offset` and `error_code`, which effectively come from
 /// [`PartitionProduceResponse`].
-pub(crate) struct ProducePartitionResponseFuture {
+#[derive(Debug, Clone)]
+pub struct ProducePartitionResponseFuture {


is still need to be public?

no, I reverted it.

sehz · 2025-01-20T19:03:20Z

crates/fluvio/src/producer/partition_producer.rs

+                    created_at: batch_metadata.created_at,
+                    metadata: batch_metadata.clone(),
+                };
+                callback.finished(event).await.unwrap();


this should be trigger when flush happen not wait callback is waiting. this will block sender

sehz · 2025-01-20T19:04:12Z

crates/fluvio/src/producer/record.rs

@@ -34,7 +35,7 @@ impl RecordMetadata {
 }

 /// Possible states of a batch in the accumulator
-pub(crate) enum BatchMetadataState {
+pub enum BatchMetadataState {


why this need to be make it public?

I reverted it.

sehz · 2025-01-20T19:04:37Z

crates/fluvio/src/producer/record.rs

        }
    }

    /// Wait for the base offset of the batch. This is the offset of the first
    /// record in the batch and it is known once the batch is sent to the server.
-    pub(crate) async fn base_offset(&self) -> Result<Offset> {
+    pub async fn base_offset(&self) -> Result<Offset> {


this should not be change into public

I reverted it.

sehz · 2025-01-20T23:56:57Z

crates/fluvio/src/producer/record.rs

@@ -43,14 +44,16 @@ pub(crate) enum BatchMetadataState {
    Failed(ProducerError),
 }

-pub(crate) struct BatchMetadata {
-    state: RwLock<BatchMetadataState>,
+pub struct BatchMetadata {


is still need to be public?

no, revert it!

crates/fluvio/src/producer/accumulator.rs

sehz · 2025-01-21T00:24:35Z

crates/fluvio/src/producer/partition_producer.rs

+                let elapsed = created_at.elapsed();
+                let event = ProduceCompletionBatchEvent {
+                    created_at,
+                    topic: self.replica.topic.clone(),


this is expensive (cloning string)

we're not using it, removed!

sehz

LGTM. Nice job

fraidev force-pushed the fluvio_producer_callback branch 13 times, most recently from 21775b3 to a283846 Compare January 20, 2025 04:08

fraidev marked this pull request as ready for review January 20, 2025 04:08

fraidev requested a review from sehz January 20, 2025 04:13

fraidev force-pushed the fluvio_producer_callback branch 2 times, most recently from 530de55 to c182387 Compare January 20, 2025 04:14

sehz reviewed Jan 20, 2025

View reviewed changes

crates/fluvio/src/producer/accumulator.rs Show resolved Hide resolved

sehz reviewed Jan 20, 2025

View reviewed changes

feat: add fluvio producer callback

1617509

chore: remove generic from producer callback

16c33c3

fraidev force-pushed the fluvio_producer_callback branch from eb0d036 to c53c036 Compare January 20, 2025 21:50

chore: return batch info via callback

145d6aa

fraidev force-pushed the fluvio_producer_callback branch from c53c036 to 145d6aa Compare January 20, 2025 21:55

fraidev requested a review from sehz January 20, 2025 22:00

chore: remove interval latency for now

59341e0

sehz reviewed Jan 20, 2025

View reviewed changes

sehz reviewed Jan 21, 2025

View reviewed changes

crates/fluvio/src/producer/accumulator.rs Show resolved Hide resolved

fraidev force-pushed the fluvio_producer_callback branch from e8abcb5 to 2f0836b Compare January 21, 2025 00:20

sehz requested changes Jan 21, 2025

View reviewed changes

chore: getting response from the callback concurrently

0e7380c

fraidev force-pushed the fluvio_producer_callback branch from 2f0836b to 0e7380c Compare January 21, 2025 00:29

sehz approved these changes Jan 21, 2025

View reviewed changes

fraidev added this pull request to the merge queue Jan 21, 2025

Merged via the queue into infinyon:master with commit 3853537 Jan 21, 2025
102 checks passed

fraidev deleted the fluvio_producer_callback branch January 21, 2025 01:25

This was referenced Jan 22, 2025

produce callback for flushing #4341

Closed

CI to run producer tests on WASM #4369

Open

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: add fluvio producer callback #4340

feat: add fluvio producer callback #4340

fraidev commented Jan 19, 2025 •

edited

Loading

sehz Jan 20, 2025

fraidev Jan 20, 2025

fraidev Jan 20, 2025

sehz Jan 20, 2025

fraidev Jan 20, 2025

sehz Jan 20, 2025

fraidev Jan 20, 2025

fraidev Jan 20, 2025 •

edited

Loading

sehz Jan 20, 2025

sehz Jan 20, 2025

fraidev Jan 20, 2025

fraidev Jan 20, 2025 •

edited

Loading

sehz Jan 20, 2025

fraidev Jan 20, 2025

sehz Jan 20, 2025

fraidev Jan 20, 2025

sehz Jan 20, 2025

fraidev Jan 20, 2025

sehz Jan 20, 2025

sehz Jan 20, 2025

fraidev Jan 20, 2025

sehz Jan 20, 2025

fraidev Jan 20, 2025

sehz Jan 20, 2025

fraidev Jan 21, 2025

sehz Jan 21, 2025

fraidev Jan 21, 2025

sehz left a comment

feat: add fluvio producer callback #4340

feat: add fluvio producer callback #4340

Conversation

fraidev commented Jan 19, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

fraidev Jan 20, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

fraidev Jan 20, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sehz left a comment

Choose a reason for hiding this comment

fraidev commented Jan 19, 2025 •

edited

Loading

fraidev Jan 20, 2025 •

edited

Loading

fraidev Jan 20, 2025 •

edited

Loading