fix: Sourcer - Change pending/partitions handlers of Source to async (#301)

Signed-off-by: Sreekanth <prsreekanth920@gmail.com>
Signed-off-by: Vigith Maurice <vigith@gmail.com>
Co-authored-by: Vigith Maurice <vigith@gmail.com>

Author: BulkBeing

packages/pynumaflow/pynumaflow/sourcer/_dtypes.py

@@ -273,12 +273,6 @@ class Sourcer(metaclass=ABCMeta):
     which will be exposed over an gRPC server.
     """
 
-    def __call__(self, *args, **kwargs):
-        """
-        Allow to call handler function directly if class instance is sent
-        """
-        return self.handler(*args, **kwargs)
-
     @abstractmethod
     async def read_handler(self, datum: ReadRequest, output: NonBlockingIterator):
         """
@@ -290,28 +284,28 @@ async def read_handler(self, datum: ReadRequest, output: NonBlockingIterator):
         pass
 
     @abstractmethod
-    def ack_handler(self, ack_request: AckRequest):
+    async def ack_handler(self, ack_request: AckRequest):
         """
         The ack handler is used to acknowledge the offsets that have been read
         """
         pass
 
     @abstractmethod
-    def nack_handler(self, nack_request: NackRequest):
+    async def nack_handler(self, nack_request: NackRequest):
         """
         The nack handler is used to negatively acknowledge the offsets on the source
         """
         pass
 
     @abstractmethod
-    def pending_handler(self) -> PendingResponse:
+    async def pending_handler(self) -> PendingResponse:
         """
         The simple source always returns zero to indicate there is no pending record.
         """
         pass
 
     @abstractmethod
-    def partitions_handler(self) -> PartitionsResponse:
+    async def partitions_handler(self) -> PartitionsResponse:
         """
         The simple source always returns zero to indicate there is no pending record.
         """

packages/pynumaflow/pynumaflow/sourcer/servicer/async_servicer.py

@@ -213,7 +213,7 @@ async def PendingFn(
         except BaseException as err:
             _LOGGER.critical("PendingFn Error", exc_info=True)
             await handle_async_error(context, err, ERR_UDF_EXCEPTION_STRING)
-            return
+            raise
         resp = source_pb2.PendingResponse.Result(count=count.count)
         return source_pb2.PendingResponse(result=resp)
 
@@ -228,7 +228,7 @@ async def PartitionsFn(
         except BaseException as err:
             _LOGGER.critical("PartitionsFn Error", exc_info=True)
             await handle_async_error(context, err, ERR_UDF_EXCEPTION_STRING)
-            return
+            raise
         resp = source_pb2.PartitionsResponse.Result(partitions=partitions.partitions)
         return source_pb2.PartitionsResponse(result=resp)