feat: implement Source trait and use it for user-defined source #2114

vigith · 2024-10-02T18:25:32Z

user-defined source over Source Trait

Signed-off-by: Vigith Maurice <[email protected]>

codecov · 2024-10-02T18:31:48Z

Codecov Report

Attention: Patch coverage is 96.08696% with 9 lines in your changes missing coverage. Please review.

Project coverage is 63.96%. Comparing base (3dbed43) to head (d99e1dd).
Report is 1 commits behind head on main.

Files with missing lines	Patch %	Lines
rust/numaflow-core/src/source/user_defined.rs	89.36%	5 Missing ⚠️
rust/numaflow-core/src/shared/server_info.rs	98.49%	2 Missing ⚠️
rust/numaflow-core/src/monovertex/forwarder.rs	95.00%	1 Missing ⚠️
rust/numaflow-core/src/reader.rs	0.00%	1 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #2114      +/-   ##
==========================================
- Coverage   64.25%   63.96%   -0.30%     
==========================================
  Files         324      325       +1     
  Lines       30650    31186     +536     
==========================================
+ Hits        19695    19947     +252     
- Misses       9913    10200     +287     
+ Partials     1042     1039       -3

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

KeranYang · 2024-10-03T00:16:20Z

rust/numaflow-core/src/message.rs

 use crate::monovertex::source_pb::{read_response, AckRequest};
 use crate::monovertex::sourcetransform_pb::SourceTransformRequest;
+use crate::monovertex::{source_pb, sourcetransform_pb};


Just curious, I saw a lot of formatting changes in the PR, how did you generate those? asking because I ran cargo fmt in my last PR and it didn't generate your changes. is it because we are using different formatting scheme?

i do "cargo fmt" right before committing.

KeranYang · 2024-10-03T00:44:56Z

rust/numaflow-core/src/monovertex/metrics.rs

@@ -365,8 +366,8 @@ struct TimestampedPending {
 /// `LagReader` is responsible for periodically checking the lag of the source client


nit: please update comments

absolutely 😄

Signed-off-by: Vigith Maurice <[email protected]>

yhl25 · 2024-10-03T03:24:36Z

rust/numaflow-core/src/monovertex.rs

+    let (source_reader, lag_reader) = new_source(source_grpc_client.clone(), 500, 100).await?;
+


Can you add a comment about the numbers 500 and 100? use constants.

yhl25 · 2024-10-03T03:27:59Z

rust/numaflow-core/src/source/user_defined.rs


-        let mut source = Source::new(client)
+        let (mut source, mut lagreader) = new_source(client, 5, 1000)
            .await


this is confusing, method name is new_source but its creating a source as well as lag_reader

yhl25 · 2024-10-03T03:30:11Z

rust/numaflow-core/src/monovertex/forwarder.rs

+            500,
+            100,


better to make it a constant, since its used in multiple places

Can we fetch it from the config?

yhl25 · 2024-10-03T03:32:39Z

rust/numaflow-core/src/monovertex.rs

    // start the lag reader to publish lag metrics
-    let mut lag_reader = utils::create_lag_reader(source_grpc_client.clone()).await;
+    let mut lag_reader = utils::create_lag_reader(lag_reader).await;
    lag_reader.start().await;


we pass lag_reader inside create_lag_reader method, which is confusing

yhl25 · 2024-10-03T03:35:14Z

rust/numaflow-core/src/monovertex/metrics.rs

+pub(crate) struct PendingReader<T> {
+    lag_reader: T,


There is a lag_reader inside PendingReader which is confusing.

lag_reader is the trait. any recommendation?

rust/numaflow-core/src/source/user_defined.rs

Signed-off-by: Vigith Maurice <[email protected]> Co-authored-by: Yashash H L <[email protected]>

rust/numaflow-core/src/source/user_defined.rs

BulkBeing · 2024-10-03T03:51:13Z

rust/numaflow-core/src/source/user_defined.rs

@@ -117,7 +143,7 @@ impl Source {
            .await
            .map_err(|e| SourceError(e.to_string()))?;

-        let mut messages = Vec::with_capacity(num_records as usize);
+        let mut messages = Vec::with_capacity(self.num_records as usize);


self.num_records is usize

Signed-off-by: Vigith Maurice <[email protected]> Co-authored-by: Sreekanth <[email protected]>

BulkBeing · 2024-10-03T03:53:31Z

rust/numaflow-core/src/source/user_defined.rs

+    let ud_src = UserDefinedSource::new(client.clone(), num_records, timeout_in_ms)
+        .await
+        .map_err(|e| e)?;


Suggested change

let ud_src = UserDefinedSource::new(client.clone(), num_records, timeout_in_ms)

.await

.map_err(|e| e)?;

let ud_src = UserDefinedSource::new(client.clone(), num_records, timeout_in_ms).await?;

Signed-off-by: Yashash H L <[email protected]>

…ce-trait

Signed-off-by: Yashash H L <[email protected]>

…proj#2114) Signed-off-by: Vigith Maurice <[email protected]> Signed-off-by: Yashash H L <[email protected]> Co-authored-by: Yashash H L <[email protected]> Co-authored-by: Sreekanth <[email protected]>

vigith added 4 commits October 2, 2024 08:37

feat: trait for Source

ec7c7c1

Signed-off-by: Vigith Maurice <[email protected]>

feat: introduce lag-reader

acb5ca4

Signed-off-by: Vigith Maurice <[email protected]>

feat: use lag-reader trait for pending

3a84b0b

Signed-off-by: Vigith Maurice <[email protected]>

chore: remove futures

7380b0c

Signed-off-by: Vigith Maurice <[email protected]>

vigith mentioned this pull request Oct 2, 2024

Implement Tickgen for MonoVertex #2113

Closed

vigith requested review from BulkBeing and yhl25 October 2, 2024 18:26

vigith self-assigned this Oct 2, 2024

KeranYang reviewed Oct 3, 2024

View reviewed changes

vigith added 2 commits October 2, 2024 18:33

feat: make monovertex::forwarder generic over source::Source

aa2d300

Signed-off-by: Vigith Maurice <[email protected]>

chore: cargo fmt

bdc5d41

Signed-off-by: Vigith Maurice <[email protected]>

vigith marked this pull request as ready for review October 3, 2024 01:57

vigith requested a review from whynowy as a code owner October 3, 2024 01:57

fix: comments

854ecbe

Signed-off-by: Vigith Maurice <[email protected]>

yhl25 reviewed Oct 3, 2024

View reviewed changes

chore: review

7a65eb1

Signed-off-by: Vigith Maurice <[email protected]> Co-authored-by: Yashash H L <[email protected]>

BulkBeing reviewed Oct 3, 2024

View reviewed changes

rust/numaflow-core/src/source/user_defined.rs Outdated Show resolved Hide resolved

BulkBeing reviewed Oct 3, 2024

View reviewed changes

chore: review

8ca2f72

Signed-off-by: Vigith Maurice <[email protected]> Co-authored-by: Sreekanth <[email protected]>

BulkBeing reviewed Oct 3, 2024

View reviewed changes

yhl25 added 2 commits October 3, 2024 09:41

minor changes

1518a49

Signed-off-by: Yashash H L <[email protected]>

Merge branch 'source-trait' of github.com:numaproj/numaflow into sour…

10eb7ba

…ce-trait

yhl25 approved these changes Oct 3, 2024

View reviewed changes

yhl25 added 2 commits October 3, 2024 10:02

add new_source

5cf88a4

Signed-off-by: Yashash H L <[email protected]>

rename

d99e1dd

Signed-off-by: Yashash H L <[email protected]>

yhl25 merged commit dc25c4d into main Oct 3, 2024
28 checks passed

yhl25 deleted the source-trait branch October 3, 2024 06:44

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: implement Source trait and use it for user-defined source #2114

feat: implement Source trait and use it for user-defined source #2114

vigith commented Oct 2, 2024

codecov bot commented Oct 2, 2024 •

edited

Loading

KeranYang Oct 3, 2024

vigith Oct 3, 2024

KeranYang Oct 3, 2024

vigith Oct 3, 2024

yhl25 Oct 3, 2024

yhl25 Oct 3, 2024 •

edited

Loading

yhl25 Oct 3, 2024

yhl25 Oct 3, 2024

yhl25 Oct 3, 2024

yhl25 Oct 3, 2024

vigith Oct 3, 2024

BulkBeing Oct 3, 2024

BulkBeing Oct 3, 2024

		@@ -365,8 +366,8 @@ struct TimestampedPending {
		/// `LagReader` is responsible for periodically checking the lag of the source client

		let (source_reader, lag_reader) = new_source(source_grpc_client.clone(), 500, 100).await?;

feat: implement Source trait and use it for user-defined source #2114

feat: implement Source trait and use it for user-defined source #2114

Conversation

vigith commented Oct 2, 2024

codecov bot commented Oct 2, 2024 • edited Loading

Codecov Report

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

yhl25 Oct 3, 2024 • 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

codecov bot commented Oct 2, 2024 •

edited

Loading

yhl25 Oct 3, 2024 •

edited

Loading