fix: tests and lint

Signed-off-by: srao12 <Shrivardhan_Rao@intuit.com>

Author: srao12

examples/accumulator/streamsorter/Dockerfile

@@ -9,7 +9,6 @@ WORKDIR $PYSETUP_PATH
 # Copy only core dependency files first for better caching
 COPY pyproject.toml poetry.lock README.md ./
 COPY pynumaflow/ ./pynumaflow/
-RUN echo "Simulating long build step..." && sleep 20
 RUN apt-get update && apt-get install --no-install-recommends -y \
         curl wget build-essential git \
     && apt-get clean && rm -rf /var/lib/apt/lists/* \

examples/accumulator/streamsorter/example.py

@@ -29,9 +29,11 @@ async def handler(
     ):
         _LOGGER.info("StreamSorter handler started")
         async for datum in datums:
-            _LOGGER.info(f"Received datum with event time: {datum.event_time}")
-            _LOGGER.info(f"Current latest watermark: {self.latest_wm}")
-            _LOGGER.info(f"Datum watermark: {datum.watermark}")
+            _LOGGER.info(
+                f"Received datum with event time: {datum.event_time}, "
+                f"Current latest watermark: {self.latest_wm}, "
+                f"Datum watermark: {datum.watermark}"
+            )
 
             # If watermark has moved forward
             if datum.watermark and datum.watermark > self.latest_wm:
@@ -68,7 +70,7 @@ async def flush_buffer(self, output: NonBlockingIterator):
     invoke = os.getenv("INVOKE", "class")
     grpc_server = None
     if invoke == "class":
-        # Here we are using the class instance as the reducer_instance
+        # Here we are using the class instance as the accumulator_instance
         # which will be used to invoke the handler function.
         # We are passing the init_args for the class instance.
         grpc_server = AccumulatorAsyncServer(StreamSorter)

examples/map/even_odd/Dockerfile

@@ -9,7 +9,6 @@ WORKDIR $PYSETUP_PATH
 # Copy only core dependency files first for better caching
 COPY pyproject.toml poetry.lock README.md ./
 COPY pynumaflow/ ./pynumaflow/
-RUN echo "Simulating long build step..." && sleep 20
 RUN apt-get update && apt-get install --no-install-recommends -y \
         curl wget build-essential git \
     && apt-get clean && rm -rf /var/lib/apt/lists/* \

pynumaflow/accumulator/_dtypes.py

@@ -32,7 +32,7 @@ class Datum:
         event_time: the event time of the event.
         watermark: the watermark of the event.
     >>> # Example usage
-    >>> from pynumaflow.reducer import Datum
+    >>> from pynumaflow.accumulator import Datum
     >>> from datetime import datetime, timezone
     >>> payload = bytes("test_mock_message", encoding="utf-8")
     >>> t1 = datetime.fromtimestamp(1662998400, timezone.utc)
@@ -419,8 +419,8 @@ class _AccumulatorBuilderClass:
 
     Args:
         accumulator_class: the Accumulator class to be used for Accumulator UDF
-        args: the arguments to be passed to the reducer class
-        kwargs: the keyword arguments to be passed to the reducer class
+        args: the arguments to be passed to the accumulator class
+        kwargs: the keyword arguments to be passed to the accumulator class
     """
 
     def __init__(self, accumulator_class: type[Accumulator], args: tuple, kwargs: dict):
@@ -430,10 +430,10 @@ def __init__(self, accumulator_class: type[Accumulator], args: tuple, kwargs: di
 
     def create(self) -> Accumulator:
         """
-        Create a new ReduceStreamer instance.
+        Create a new Accumulator instance.
         """
         return self._accumulator_class(*self._args, **self._kwargs)
 
 
-# AccumulatorStreamCallable is a callable which can be used as a handler for the Reduce UDF.
+# AccumulatorStreamCallable is a callable which can be used as a handler for the Accumulator UDF.
 AccumulatorStreamCallable = Union[AccumulatorAsyncCallable, type[Accumulator]]

pynumaflow/accumulator/async_server.py

@@ -62,8 +62,8 @@ class AccumulatorAsyncServer(NumaflowServer):
     A new servicer instance is created and attached to the server.
     The server instance is returned.
     Args:
-        accumulator_instance: The reducer instance to be used for
-                Reduce Streaming UDF
+        accumulator_instance: The accumulator instance to be used for
+                Accumulator UDF
         init_args: The arguments to be passed to the accumulator_handler
         init_kwargs: The keyword arguments to be passed to the
             accumulator_handler
@@ -78,7 +78,7 @@ class AccumulatorAsyncServer(NumaflowServer):
         from pynumaflow.accumulator import Messages, Message, Datum, Metadata,
         AccumulatorAsyncServer, Accumulator
 
-        class ReduceCounter(Accumulator):
+        class StreamSorter(Accumulator):
             def __init__(self, counter):
                 self.counter = counter
 
@@ -117,13 +117,13 @@ async def reduce_handler(
         if __name__ == "__main__":
             invoke = os.getenv("INVOKE", "func_handler")
             if invoke == "class":
-                # Here we are using the class instance as the reducer_instance
+                # Here we are using the class instance as the accumulator_instance
                 # which will be used to invoke the handler function.
                 # We are passing the init_args for the class instance.
-                grpc_server = AccumulatorAsyncServer(ReduceCounter, init_args=(0,))
+                grpc_server = AccumulatorAsyncServer(StreamSorter, init_args=(0,))
             else:
-                # Here we are using the handler function directly as the reducer_instance.
-                grpc_server = AccumulatorAsyncServer(reduce_handler)
+                # Here we are using the handler function directly as the accumulator_instance.
+                grpc_server = AccumulatorAsyncServer(accumulator_handler)
             grpc_server.start()
 
     """

pynumaflow/accumulator/servicer/async_servicer.py

@@ -48,7 +48,7 @@ async def datum_generator(
 
 class AsyncAccumulatorServicer(accumulator_pb2_grpc.AccumulatorServicer):
     """
-    This class is used to create a new grpc Reduce servicer instance.
+    This class is used to create a new grpc Accumulator servicer instance.
     Provides the functionality for the required rpc methods.
     """
 
@@ -90,17 +90,17 @@ async def AccumulateFn(
         # Start the consumer task where we read from the result queue
         # and send the results to the client
         # The task manager can write the following to the result queue:
-        # 1. A accumulator_pb2.ReduceResponse message
-        # This is the result of the reduce function, it contains the window and the
-        # result of the reduce function
-        # The result of the reduce function is a accumulator_pb2.ReduceResponse message and can be
-        # directly sent to the client
+        # 1. A accumulator_pb2.AccumulatorResponse message
+        # This is the result of the accumulator function, it contains the window and the
+        # result of the accumulator function
+        # The result of the accumulator function is a accumulator_pb2.AccumulatorResponse message
+        #  and can be directly sent to the client
         #
         # 2. An Exception
-        # Any exceptions that occur during the processing reduce function tasks are
+        # Any exceptions that occur during the processing accumulator function tasks are
         # sent to the result queue. We then forward these exception to the client
         #
-        # 3. A accumulator_pb2.ReduceResponse message with EOF=True
+        # 3. A accumulator_pb2.AccumulatorResponse message with EOF=True
         # This is a special message that indicates the end of the processing for a window
         # When we get this message, we send an EOF message to the client
         try:

pynumaflow/accumulator/servicer/task_manager.py

@@ -23,7 +23,7 @@
 def build_unique_key_name(keys):
     """
     Builds a unique key name for the given keys and window.
-    The key name is used to identify the Reduce task.
+    The key name is used to identify the Accumulator task.
     The format is: start_time:end_time:key1:key2:...
     """
     return f"{DELIMITER.join(keys)}"
@@ -32,21 +32,21 @@ def build_unique_key_name(keys):
 def build_window_hash(window):
     """
     Builds a hash for the given window.
-    The hash is used to identify the Reduce Window
+    The hash is used to identify the Accumulator Window
     The format is: start_time:end_time
     """
     return f"{window.start.ToMilliseconds()}:{window.end.ToMilliseconds()}"
 
 
 def create_window_eof_response(window):
-    """Create a Reduce response with EOF=True for a given window"""
-    return accumulator_pb2.ReduceResponse(window=window, EOF=True)
+    """Create a Accumulator response with EOF=True for a given window"""
+    return accumulator_pb2.AccumulatorResponse(window=window, EOF=True)
 
 
 class TaskManager:
     """
-    TaskManager is responsible for managing the Reduce tasks.
-    It is created whenever a new reduce operation is requested.
+    TaskManager is responsible for managing the Accumulator tasks.
+    It is created whenever a new accumulator operation is requested.
     """
 
     def __init__(self, handler: Union[AccumulatorAsyncCallable, _AccumulatorBuilderClass]):
@@ -56,13 +56,13 @@ def __init__(self, handler: Union[AccumulatorAsyncCallable, _AccumulatorBuilderC
         # Event loop only keeps a weak reference, which can cause it to
         # get lost during execution.
         self.background_tasks = set()
-        # Handler for the reduce operation
+        # Handler for the accumulator operation
         self.__accumulator_handler = handler
-        # Queue to store the results of the reduce operation
+        # Queue to store the results of the accumulator operation
         # This queue is used to send the results to the client
-        # once the reduce operation is completed.
+        # once the accumulator operation is completed.
         # This queue is also used to send the error/exceptions to the client
-        # if the reduce operation fails.
+        # if the accumulator operation fails.
         self.global_result_queue = NonBlockingIterator()
         # EOF response counting to ensure proper termination
         self._expected_eof_count = 0
@@ -87,14 +87,14 @@ def get_unique_windows(self):
 
     def get_tasks(self):
         """
-        Returns the list of reduce tasks that are
+        Returns the list of accumulator tasks that are
         currently being processed
         """
         return list(self.tasks.values())
 
     async def stream_send_eof(self):
         """
-        Sends EOF to input streams of all the Reduce
+        Sends EOF to input streams of all the Accumulator
         tasks that are currently being processed.
         This is called when the input grpc stream is closed.
         """
@@ -136,7 +136,7 @@ async def create_task(self, req):
             riter = niter.read_iterator()
             # Create a new result queue for the current task
             # We create a new result queue for each task, so that
-            # the results of the reduce operation can be sent to the
+            # the results of the accumulator operation can be sent to the
             # the global result queue, which in turn sends the results
             # to the client.
             res_queue = NonBlockingIterator()
@@ -153,7 +153,7 @@ async def create_task(self, req):
             consumer.add_done_callback(self.clean_background)
 
             # Create a new task for the accumulator operation, this will invoke the
-            # Reduce handler with the given keys, request iterator, and window.
+            # Accumulator handler with the given keys, request iterator, and window.
             task = asyncio.create_task(self.__invoke_accumulator(riter, res_queue))
             # Save a reference to the result of this function, to avoid a
             # task disappearing mid-execution.
@@ -165,7 +165,7 @@ async def create_task(self, req):
                 task, niter, keys, res_queue, consumer, datetime.fromtimestamp(-1)
             )
 
-            # Save the result of the reduce operation to the task list
+            # Save the result of the accumulator operation to the task list
             self.tasks[unified_key] = curr_task
 
             # Increment expected EOF count since we created a new task
@@ -195,23 +195,23 @@ async def __invoke_accumulator(
         output: NonBlockingIterator,
     ):
         """
-        Invokes the UDF reduce handler with the given keys,
+        Invokes the UDF accumulator handler with the given keys,
         request iterator, and window. Returns the result of the
-        reduce operation.
+        accumulator operation.
         """
         new_instance = self.__accumulator_handler
 
         # If the accumulator handler is a class instance, create a new instance of it.
         # It is required for a new key to be processed by a
-        # new instance of the reducer for a given window
+        # new instance of the accumulator for a given window
         # Otherwise the function handler can be called directly
         if isinstance(self.__accumulator_handler, _AccumulatorBuilderClass):
             new_instance = self.__accumulator_handler.create()
         try:
             _ = await new_instance(request_iterator, output)
             # send EOF to the output stream
             await output.put(STREAM_EOF)
-        # If there is an error in the reduce operation, log and
+        # If there is an error in the accumulator operation, log and
         # then send the error to the result queue
         except BaseException as err:
             _LOGGER.critical("panic inside accumulator handle", exc_info=True)
@@ -242,7 +242,7 @@ async def process_input_stream(
                 else:
                     _LOGGER.debug(f"No operation matched for request: {request}", exc_info=True)
 
-        # If there is an error in the reduce operation, log and
+        # If there is an error in the accumulator operation, log and
         # then send the error to the result queue
         except BaseException as e:
             err_msg = f"Accumulator Error: {repr(e)}"
@@ -257,7 +257,7 @@ async def process_input_stream(
             # respective iterators.
             await self.stream_send_eof()
 
-            # get the list of reduce tasks that are currently being processed
+            # get the list of accumulator tasks that are currently being processed
             # iterate through the tasks and wait for them to complete
             for task in self.get_tasks():
                 # Once this is done, we know that the task has written all the results
@@ -281,15 +281,15 @@ async def process_input_stream(
             # Now send STREAM_EOF to terminate the global result queue iterator
             await self.global_result_queue.put(STREAM_EOF)
         except BaseException as e:
-            err_msg = f"Reduce Streaming Error: {repr(e)}"
+            err_msg = f"Accumulator Streaming Error: {repr(e)}"
             _LOGGER.critical(err_msg, exc_info=True)
             await self.global_result_queue.put(e)
 
     async def write_to_global_queue(
         self, input_queue: NonBlockingIterator, output_queue: NonBlockingIterator, unified_key: str
     ):
         """
-        This task is for given Reduce task.
+        This task is for given Accumulator task.
         This would from the local result queue for the task and then write
         to the global result queue
         """

tests/accumulator/test_async_accumulator.py

@@ -270,6 +270,60 @@ def test_max_threads(self):
         server = AccumulatorAsyncServer(accumulator_instance=ExampleClass)
         self.assertEqual(server.max_threads, 4)
 
+    def test_server_info_file_path_handling(self):
+        """Test AccumulatorAsyncServer with custom server info file path."""
+
+        server = AccumulatorAsyncServer(
+            ExampleClass, init_args=(0,), server_info_file="/custom/path/server_info.json"
+        )
+
+        self.assertEqual(server.server_info_file, "/custom/path/server_info.json")
+
+    def test_init_kwargs_none_handling(self):
+        """Test init_kwargs None handling in AccumulatorAsyncServer."""
+
+        server = AccumulatorAsyncServer(
+            ExampleClass, init_args=(0,), init_kwargs=None  # This should be converted to {}
+        )
+
+        # Should not raise any errors and should work correctly
+        self.assertIsNotNone(server.accumulator_handler)
+
+    def test_server_with_zero_max_threads(self):
+        """Test server creation with max_threads set to 0."""
+
+        server = AccumulatorAsyncServer(ExampleClass, max_threads=0)
+
+        # Should be set to 0 (minimum)
+        self.assertEqual(server.max_threads, 0)
+
+    def test_server_with_negative_max_threads(self):
+        """Test server creation with negative max_threads."""
+
+        server = AccumulatorAsyncServer(ExampleClass, max_threads=-5)
+
+        # Should be set to -5 (the minimum function will handle this)
+        self.assertEqual(server.max_threads, -5)
+
+    def test_server_start_method_logging(self):
+        """Test server start method includes proper logging."""
+        from unittest.mock import patch
+
+        server = AccumulatorAsyncServer(ExampleClass)
+
+        # Mock aiorun.run to prevent actual server startup
+        with patch("pynumaflow.accumulator.async_server.aiorun") as mock_aiorun, patch(
+            "pynumaflow.accumulator.async_server._LOGGER"
+        ) as mock_logger:
+            server.start()
+
+            # Verify logging was called
+            mock_logger.info.assert_called_once_with("Starting Async Accumulator Server")
+
+            # Verify aiorun.run was called with correct parameters
+            mock_aiorun.run.assert_called_once()
+            self.assertTrue(mock_aiorun.run.call_args[1]["use_uvloop"])
+
 
 if __name__ == "__main__":
     logging.basicConfig(level=logging.DEBUG)