chore: avoid creating window message again (#2670)

Signed-off-by: Yashash H L <yashashhl25@gmail.com>

Author: yhl25

rust/numaflow-core/src/reduce/reducer/aligned/reducer.rs

@@ -226,27 +226,27 @@ impl AlignedReduceActor {
     /// Runs the actor, listening for messages and multiplexing them to the reduce tasks.
     async fn run(mut self) {
         while let Some(msg) = self.receiver.recv().await {
-            self.handle_window_message(msg.window, msg.operation).await;
+            self.handle_window_message(msg).await;
         }
         self.wait_for_all_tasks().await;
     }
 
     /// Handle a window message based on its operation type
-    async fn handle_window_message(&mut self, window: Window, operation: WindowOperation) {
-        let window_id = window.pnf_slot();
-        match operation {
-            WindowOperation::Open(msg) => self.window_open(window, window_id, msg).await,
-            WindowOperation::Append(msg) => self.window_append(window, window_id, msg).await,
-            WindowOperation::Close => self.window_close(window_id).await,
+    async fn handle_window_message(&mut self, window_msg: AlignedWindowMessage) {
+        match &window_msg.operation {
+            WindowOperation::Open(_) => self.window_open(window_msg).await,
+            WindowOperation::Append(_) => self.window_append(window_msg).await,
+            WindowOperation::Close => self.window_close(window_msg).await,
         }
     }
 
     /// Creates a new reduce task for the window and sends the initial Open command with the
     /// first message.
-    async fn window_open(&mut self, window: Window, window_id: Bytes, msg: Message) {
+    async fn window_open(&mut self, window_msg: AlignedWindowMessage) {
         // Create a new channel for this window's messages
         let (message_tx, message_rx) = mpsc::channel(100);
         let message_stream = ReceiverStream::new(message_rx);
+        let window = window_msg.window.clone();
 
         // Create a ReduceTask
         let reduce_task = ReduceTask::new(
@@ -258,20 +258,14 @@ impl AlignedReduceActor {
             self.window_manager.clone(),
         );
 
-        // Create the initial window message
-        let window_msg = AlignedWindowMessage {
-            operation: WindowOperation::Open(msg),
-            window: window.clone(),
-        };
-
         // start the reduce task and store the handle and the sender so that we can send messages
         // and wait for it to complete.
         let task_handle = reduce_task
             .start(message_stream, self.cln_token.clone())
             .await;
 
         self.active_streams.insert(
-            window_id,
+            window.pnf_slot(),
             ActiveStream {
                 message_tx: message_tx.clone(),
                 task_handle,
@@ -283,24 +277,21 @@ impl AlignedReduceActor {
     }
 
     /// sends the message to the reduce task for the window.
-    async fn window_append(&mut self, window: Window, window_id: Bytes, msg: Message) {
+    async fn window_append(&mut self, window_msg: AlignedWindowMessage) {
+        let window = window_msg.window.clone();
+        let window_id = window.pnf_slot();
+
         // Get the existing stream or log error if not found create a new one.
         let Some(active_stream) = self.active_streams.get(&window_id) else {
             // windows may not be found during replay, because the windower doesn't send the open
             // message for the active windows that got replayed, hence we create a new one.
             // this happens because of out-of-order messages and we have to ensure that the (t+1)th
             // message is sent to the window that could be created by (t)th message iff (t+1)th message
             // belongs to that window created by (t)th message.
-            self.window_open(window, window_id, msg).await;
+            self.window_open(window_msg).await;
             return;
         };
 
-        // Create the append window message
-        let window_msg = AlignedWindowMessage {
-            operation: WindowOperation::Append(msg),
-            window,
-        };
-
         // Send the append message
         active_stream
             .message_tx
@@ -310,7 +301,9 @@ impl AlignedReduceActor {
     }
 
     /// Closes the reduce task for the window.
-    async fn window_close(&mut self, window_id: Bytes) {
+    async fn window_close(&mut self, window_msg: AlignedWindowMessage) {
+        let window_id = window_msg.window.pnf_slot();
+
         // Get the existing stream or log error if not found
         let Some(active_stream) = self.active_streams.remove(&window_id) else {
             error!("No active stream found for window {:?}", window_id);

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

@@ -38,7 +38,6 @@ use numaflow_pb::clients::source::source_client::SourceClient;
 use numaflow_pb::clients::sourcetransformer::source_transform_client::SourceTransformClient;
 use numaflow_sqs::sink::SqsSinkBuilder;
 use tokio_util::sync::CancellationToken;
-use tracing::info;
 
 /// Creates a sink writer based on the configuration
 pub(crate) async fn create_sink_writer(