feat: Built-in Pulsar source #2237

BulkBeing · 2024-11-22T14:37:24Z

Status: Tested by running a Pipeline and a Monovertex with Pulsar as built-in source

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

Signed-off-by: Sreekanth <[email protected]>

codecov · 2024-11-22T14:46:11Z

Codecov Report

Attention: Patch coverage is 73.85621% with 120 lines in your changes missing coverage. Please review.

Project coverage is 63.95%. Comparing base (c5afc90) to head (8d4d4e1).
Report is 4 commits behind head on main.

Files with missing lines	Patch %	Lines
rust/numaflow-extns/pulsar/src/source.rs	76.53%	42 Missing ⚠️
rust/numaflow-core/src/monovertex.rs	37.50%	15 Missing ⚠️
rust/numaflow-core/src/source/pulsar.rs	86.60%	15 Missing ⚠️
rust/numaflow-core/src/config/components.rs	76.66%	14 Missing ⚠️
rust/numaflow-core/src/source.rs	0.00%	12 Missing ⚠️
rust/numaflow-core/src/pipeline.rs	0.00%	8 Missing ⚠️
rust/numaflow-core/src/shared/utils.rs	0.00%	6 Missing ⚠️
rust/numaflow-extns/pulsar/src/lib.rs	0.00%	6 Missing ⚠️
rust/numaflow-core/src/message.rs	0.00%	2 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #2237      +/-   ##
==========================================
+ Coverage   63.91%   63.95%   +0.03%     
==========================================
  Files         338      342       +4     
  Lines       41085    41203     +118     
==========================================
+ Hits        26259    26350      +91     
- Misses      13756    13780      +24     
- Partials     1070     1073       +3

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

Signed-off-by: Sreekanth <[email protected]>

vigith · 2024-11-25T16:06:08Z

rust/numaflow-core/src/message.rs

+impl TryFrom<PulsarMessage> for Message {
+    type Error = Error;
+
+    fn try_from(message: PulsarMessage) -> Result<Self> {
+        let offset = Offset::Int(IntOffset::new(message.offset, 1)); // FIXME: partition id
+
+        Ok(Message {
+            keys: vec![message.key],
+            value: message.payload,
+            offset: Some(offset.clone()),
+            event_time: message.event_time,
+            id: MessageID {
+                vertex_name: get_vertex_name().to_string(),
+                offset: offset.to_string(),
+                index: 0,
+            },
+            headers: message.headers,
+        })
+    }
+}
+


move this to source/pulsar.rs

vigith · 2024-11-25T16:28:12Z

pkg/apis/numaflow/v1alpha1/pulsar_source.go

+type PulsarSource struct {
+	ServerAddr       string `json:"serverAddr" protobuf:"bytes,1,name=server_addr"`
+	Topic            string `json:"topic" protobuf:"bytes,2,name=topic"`
+	ConsumerName     string `json:"consumerName" protobuf:"bytes,3,name=consumerName"`
+	SubscriptionName string `json:"subscriptionName" protobuf:"bytes,4,name=subscriptionName"`
+	MaxUnack         uint32 `json:"maxUnack,omitempty" protobuf:"bytes,5,opt,name=maxUnack"`
+}


please add TLS and Auth too, if you prefer you can do that in a follow up PR.

Signed-off-by: Sreekanth <[email protected]>

vigith · 2024-11-29T15:16:56Z

rust/numaflow-core/src/config/components.rs

+            if source.udsource.is_some() {
+                return Ok(SourceType::UserDefined(UserDefinedConfig::default()));
+            }


Suggested change

if source.udsource.is_some() {

return Ok(SourceType::UserDefined(UserDefinedConfig::default()));

}

if source.udsource.is_some().take() {

return Ok(SourceType::UserDefined(UserDefinedConfig::default()));

}

Option.take() returns the inner value as an option. We can't use it in if statement.

vigith · 2024-11-29T15:18:28Z

rust/numaflow-core/src/error.rs

+impl From<numaflow_pulsar::Error> for Error {
+    fn from(value: numaflow_pulsar::Error) -> Self {
+        match value {
+            numaflow_pulsar::Error::Pulsar(e) => Error::Source(e.to_string()),
+            numaflow_pulsar::Error::UnknownOffset(_) => Error::Source(value.to_string()),
+            numaflow_pulsar::Error::AckPendingExceeded(pending) => {
+                Error::AckPendingExceeded(pending)
+            }
+            numaflow_pulsar::Error::ActorTaskTerminated(_) => {
+                Error::ActorPatternRecv(value.to_string())
+            }
+            numaflow_pulsar::Error::Other(e) => Error::Source(e),
+        }
+    }
+}


move this to `source/pulsar.rs?

vigith · 2024-11-29T15:18:59Z

rust/numaflow-core/src/message.rs

+    pub offset: u64,
+    pub partition_idx: u16,


Suggested change

pub offset: u64,

pub partition_idx: u16,

pub(crate) offset: u64,

pub(crate) partition_idx: u16,

vigith · 2024-11-29T15:20:38Z

rust/numaflow-core/src/shared/utils.rs

@@ -207,6 +207,41 @@ pub(crate) async fn create_sink_handle(
    }
 }

+/*


remove this?

vigith · 2024-11-29T15:22:09Z

rust/numaflow-core/src/source.rs

@@ -14,6 +14,7 @@ pub(crate) mod user_defined;
 ///
 /// [Generator]: https://numaflow.numaproj.io/user-guide/sources/generator/
 pub(crate) mod generator;
+pub(crate) mod pulsar;


vigith · 2024-11-29T15:27:46Z

pkg/apis/numaflow/v1alpha1/generated.proto

is there any value in committing this file? /cc @whynowy

Signed-off-by: Sreekanth <[email protected]>

vigith

please add metrics in a follow up PR

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

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

vigith and others added 2 commits November 21, 2024 20:25

feat: pulsar integration

2586b28

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

feat: Pulsar source implementation

d934976

Signed-off-by: Sreekanth <[email protected]>

BulkBeing added 7 commits November 24, 2024 06:53

Avoid serializing/deserializing pulsar offset

0c7062a

Signed-off-by: Sreekanth <[email protected]>

Conversion from Pulsar message to numaflow message

2d21a59

Signed-off-by: Sreekanth <[email protected]>

Error handling

1e13ea8

Signed-off-by: Sreekanth <[email protected]>

Code generation for Pulsar config object

17d4871

Signed-off-by: Sreekanth <[email protected]>

TTested built-in Pulsar source with Pipeline and Monovertex

6dd6a1e

Signed-off-by: Sreekanth <[email protected]>

Formatting, Upgrade Rust version to 1.82

c428d90

Signed-off-by: Sreekanth <[email protected]>

Unit test for Pulsar config parsing

1d5ee07

Signed-off-by: Sreekanth <[email protected]>

vigith reviewed Nov 25, 2024

View reviewed changes

BulkBeing added 8 commits November 27, 2024 15:49

Use rustls instead of native tls

adfc09b

Signed-off-by: Sreekanth <[email protected]>

Code re-arrangement

230ad2b

Signed-off-by: Sreekanth <[email protected]>

Option to allow self-signed certs

a14cd73

Signed-off-by: Sreekanth <[email protected]>

Tests for Pulsar source

0db2f48

Signed-off-by: Sreekanth <[email protected]>

Use docker to run Pulsar test container

5d210ab

Signed-off-by: Sreekanth <[email protected]>

Revert to rustls instead of native_tls (Pulsar dependency)

cd0a8a7

Signed-off-by: Sreekanth <[email protected]>

Ability to authenticate with Pulsar using JWT Token

68057be

Signed-off-by: Sreekanth <[email protected]>

Code generation

9b9c30b

Signed-off-by: Sreekanth <[email protected]>

vigith reviewed Nov 29, 2024

View reviewed changes

BulkBeing added 3 commits December 2, 2024 10:09

Minor refactoring based on PR review

4ddecbb

Signed-off-by: Sreekanth <[email protected]>

User Documentation for Pulsar Source

3d6c5e8

Signed-off-by: Sreekanth <[email protected]>

User doc updates

5555524

Signed-off-by: Sreekanth <[email protected]>

BulkBeing marked this pull request as ready for review December 2, 2024 06:30

BulkBeing requested a review from whynowy as a code owner December 2, 2024 06:30

vigith approved these changes Dec 2, 2024

View reviewed changes

chore: add version

8d4d4e1

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

yhl25 approved these changes Dec 3, 2024

View reviewed changes

yhl25 merged commit b31380b into main Dec 3, 2024
28 checks passed

yhl25 deleted the pulsar branch December 3, 2024 03:38

This was referenced Dec 3, 2024

Expose metrics for Pulsar source #2252

Open

Basic auth support for Pulsar source #2253

Open

vigith mentioned this pull request Dec 3, 2024

Implement Pulsar Source and Sink as builtin #2231

Open

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: Built-in Pulsar source #2237

feat: Built-in Pulsar source #2237

BulkBeing commented Nov 22, 2024 •

edited

Loading

codecov bot commented Nov 22, 2024 •

edited

Loading

vigith Nov 25, 2024

vigith Nov 25, 2024

vigith Nov 29, 2024

BulkBeing Dec 2, 2024

vigith Nov 29, 2024

vigith Nov 29, 2024

vigith Nov 29, 2024

vigith Nov 29, 2024

vigith Nov 29, 2024

vigith left a comment

@@ @@ -207,6 +207,41 @@ pub(crate) async fn create_sink_handle( @@
                   }
               }
+              /*

feat: Built-in Pulsar source #2237

feat: Built-in Pulsar source #2237

Conversation

BulkBeing commented Nov 22, 2024 • edited Loading

codecov bot commented Nov 22, 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

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

vigith left a comment

Choose a reason for hiding this comment

BulkBeing commented Nov 22, 2024 •

edited

Loading

codecov bot commented Nov 22, 2024 •

edited

Loading