feat: Idle Watermark Implementation Inside Async Data Movement for So…

…urce and ISB (#2385) Signed-off-by: Yashash H L <[email protected]> Signed-off-by: Vigith Maurice <[email protected]> Co-authored-by: Vigith Maurice <[email protected]>
numaproj · Feb 11, 2025 · 8d51c66 · 8d51c66
1 parent bf0f9db
commit 8d51c66
Show file tree

Hide file tree

Showing 44 changed files with 3,024 additions and 469 deletions.
diff --git a/pkg/shared/idlehandler/source_idlehandler.go b/pkg/shared/idlehandler/source_idlehandler.go
@@ -105,7 +105,7 @@ func (iw *SourceIdleHandler) PublishSourceIdleWatermark(partitions []int32) {
 		// in this case, we can publish the idle watermark as the last published idle watermark + the increment by value.
 		nextIdleWM = iw.lastPublishedIdleWm.Add(iw.config.IdleSource.GetIncrementBy())
 	} else {
-		// if its not -1, then we can publish the idle watermark as the computed watermark + the increment by value.
+		// if it's not -1, then we can publish the idle watermark as the computed watermark + the increment by value.
 		nextIdleWM = computedWm.Add(iw.config.IdleSource.GetIncrementBy())
 	}
 

diff --git a/pkg/sources/forward/data_forward.go b/pkg/sources/forward/data_forward.go
@@ -250,7 +250,7 @@ func (df *DataForward) forwardAChunk(ctx context.Context) error {
 		for toVertexName, toVertexBuffers := range df.toBuffers {
 			for index := range toVertexBuffers {
 				// publish idle watermark to all the source partitions owned by this reader.
-				// it is 1:1 for many (HTTP, tickgen, etc.) but for e.g., for Kafka it is 1:N and the list of partitions in the N could keep changing.
+				// it is 1:1 for many (HTTP, tick-gen, etc.) but for e.g., for Kafka it is 1:N and the list of partitions in the N could keep changing.
 				for _, sp := range df.reader.Partitions(df.ctx) {
 					if vertexPublishers, ok := df.toVertexWMPublishers[toVertexName]; ok {
 						var publisher, ok = vertexPublishers[sp]

diff --git a/rust/Cargo.lock b/rust/Cargo.lock
diff --git a/rust/extns/numaflow-pulsar/src/source.rs b/rust/extns/numaflow-pulsar/src/source.rs
@@ -238,20 +238,23 @@ pub struct PulsarSource {
     /// timeout for each batch read request
     timeout: Duration,
     actor_tx: mpsc::Sender<ConsumerActorMessage>,
+    vertex_replica: u16,
 }
 
 impl PulsarSource {
     pub async fn new(
         config: PulsarSourceConfig,
         batch_size: usize,
         timeout: Duration,
+        vertex_replica: u16,
     ) -> Result<Self> {
         let (tx, rx) = mpsc::channel(10);
         ConsumerReaderActor::start(config, rx).await?;
         Ok(Self {
             actor_tx: tx,
             batch_size,
             timeout,
+            vertex_replica,
         })
     }
 }
@@ -292,7 +295,7 @@ impl PulsarSource {
         None
     }
 
-    pub fn partitions(&self) -> Vec<u16> {
-        unimplemented!()
+    pub fn partitions_vec(&self) -> Vec<u16> {
+        vec![self.vertex_replica]
     }
 }