More documentation comments

Signed-off-by: Sreekanth <prsreekanth920@gmail.com>

Author: BulkBeing

packages/pynumaflow/pynumaflow/accumulator/async_server.py

@@ -220,9 +220,12 @@ async def _watch_for_shutdown():
         try:
             await server.wait_for_termination()
         except asyncio.CancelledError:
-            # SIGTERM received — aiorun cancels all tasks. We must stop
-            # the gRPC server explicitly so its __del__ doesn't try to
-            # schedule a coroutine on the already-closed event loop.
+            # SIGTERM received — aiorun cancels all tasks. Unlike the UDF-error
+            # path (where _watch_for_shutdown calls server.stop()), this path
+            # must stop the gRPC server explicitly. Without this, the server
+            # object is never stopped and when it is garbage-collected, its
+            # __del__ tries to schedule a cleanup coroutine on an event loop
+            # that is already closed, causing errors/warnings.
             _LOGGER.info("Received cancellation, stopping server gracefully...")
             await server.stop(NUMAFLOW_GRPC_SHUTDOWN_GRACE_PERIOD_SECONDS)
 

packages/pynumaflow/pynumaflow/batchmapper/async_server.py

@@ -155,9 +155,12 @@ async def _watch_for_shutdown():
         try:
             await server.wait_for_termination()
         except asyncio.CancelledError:
-            # SIGTERM received — aiorun cancels all tasks. We must stop
-            # the gRPC server explicitly so its __del__ doesn't try to
-            # schedule a coroutine on the already-closed event loop.
+            # SIGTERM received — aiorun cancels all tasks. Unlike the UDF-error
+            # path (where _watch_for_shutdown calls server.stop()), this path
+            # must stop the gRPC server explicitly. Without this, the server
+            # object is never stopped and when it is garbage-collected, its
+            # __del__ tries to schedule a cleanup coroutine on an event loop
+            # that is already closed, causing errors/warnings.
             _LOGGER.info("Received cancellation, stopping server gracefully...")
             await server.stop(NUMAFLOW_GRPC_SHUTDOWN_GRACE_PERIOD_SECONDS)
 

packages/pynumaflow/pynumaflow/mapper/async_server.py

@@ -153,9 +153,12 @@ async def _watch_for_shutdown():
         try:
             await server.wait_for_termination()
         except asyncio.CancelledError:
-            # SIGTERM received — aiorun cancels all tasks. We must stop
-            # the gRPC server explicitly so its __del__ doesn't try to
-            # schedule a coroutine on the already-closed event loop.
+            # SIGTERM received — aiorun cancels all tasks. Unlike the UDF-error
+            # path (where _watch_for_shutdown calls server.stop()), this path
+            # must stop the gRPC server explicitly. Without this, the server
+            # object is never stopped and when it is garbage-collected, its
+            # __del__ tries to schedule a cleanup coroutine on an event loop
+            # that is already closed, causing errors/warnings.
             _LOGGER.info("Received cancellation, stopping server gracefully...")
             await server.stop(NUMAFLOW_GRPC_SHUTDOWN_GRACE_PERIOD_SECONDS)
 

packages/pynumaflow/pynumaflow/mapstreamer/async_server.py

@@ -174,9 +174,12 @@ async def _watch_for_shutdown():
         try:
             await server.wait_for_termination()
         except asyncio.CancelledError:
-            # SIGTERM received — aiorun cancels all tasks. We must stop
-            # the gRPC server explicitly so its __del__ doesn't try to
-            # schedule a coroutine on the already-closed event loop.
+            # SIGTERM received — aiorun cancels all tasks. Unlike the UDF-error
+            # path (where _watch_for_shutdown calls server.stop()), this path
+            # must stop the gRPC server explicitly. Without this, the server
+            # object is never stopped and when it is garbage-collected, its
+            # __del__ tries to schedule a cleanup coroutine on an event loop
+            # that is already closed, causing errors/warnings.
             _LOGGER.info("Received cancellation, stopping server gracefully...")
             await server.stop(NUMAFLOW_GRPC_SHUTDOWN_GRACE_PERIOD_SECONDS)
 

packages/pynumaflow/pynumaflow/reducer/async_server.py

@@ -206,9 +206,12 @@ async def _watch_for_shutdown():
         try:
             await server.wait_for_termination()
         except asyncio.CancelledError:
-            # SIGTERM received — aiorun cancels all tasks. We must stop
-            # the gRPC server explicitly so its __del__ doesn't try to
-            # schedule a coroutine on the already-closed event loop.
+            # SIGTERM received — aiorun cancels all tasks. Unlike the UDF-error
+            # path (where _watch_for_shutdown calls server.stop()), this path
+            # must stop the gRPC server explicitly. Without this, the server
+            # object is never stopped and when it is garbage-collected, its
+            # __del__ tries to schedule a cleanup coroutine on an event loop
+            # that is already closed, causing errors/warnings.
             _LOGGER.info("Received cancellation, stopping server gracefully...")
             await server.stop(NUMAFLOW_GRPC_SHUTDOWN_GRACE_PERIOD_SECONDS)
 

packages/pynumaflow/pynumaflow/reducer/servicer/async_servicer.py

@@ -159,9 +159,7 @@ async def ReduceFn(
 
         except BaseException as e:
             _LOGGER.critical("Reduce Error", exc_info=True)
-            err_msg = f"{ERR_UDF_EXCEPTION_STRING}: {repr(e)}"
-            _LOGGER.critical(err_msg, exc_info=True)
-            update_context_err(context, e, err_msg)
+            update_context_err(context, e, f"{ERR_UDF_EXCEPTION_STRING}: {repr(e)}")
             self._error = e
             if self._shutdown_event is not None:
                 self._shutdown_event.set()

packages/pynumaflow/pynumaflow/reducestreamer/async_server.py

@@ -223,9 +223,12 @@ async def _watch_for_shutdown():
         try:
             await server.wait_for_termination()
         except asyncio.CancelledError:
-            # SIGTERM received — aiorun cancels all tasks. We must stop
-            # the gRPC server explicitly so its __del__ doesn't try to
-            # schedule a coroutine on the already-closed event loop.
+            # SIGTERM received — aiorun cancels all tasks. Unlike the UDF-error
+            # path (where _watch_for_shutdown calls server.stop()), this path
+            # must stop the gRPC server explicitly. Without this, the server
+            # object is never stopped and when it is garbage-collected, its
+            # __del__ tries to schedule a cleanup coroutine on an event loop
+            # that is already closed, causing errors/warnings.
             _LOGGER.info("Received cancellation, stopping server gracefully...")
             await server.stop(NUMAFLOW_GRPC_SHUTDOWN_GRACE_PERIOD_SECONDS)
 

packages/pynumaflow/pynumaflow/sinker/async_server.py

@@ -177,9 +177,12 @@ async def _watch_for_shutdown():
         try:
             await server.wait_for_termination()
         except asyncio.CancelledError:
-            # SIGTERM received — aiorun cancels all tasks. We must stop
-            # the gRPC server explicitly so its __del__ doesn't try to
-            # schedule a coroutine on the already-closed event loop.
+            # SIGTERM received — aiorun cancels all tasks. Unlike the UDF-error
+            # path (where _watch_for_shutdown calls server.stop()), this path
+            # must stop the gRPC server explicitly. Without this, the server
+            # object is never stopped and when it is garbage-collected, its
+            # __del__ tries to schedule a cleanup coroutine on an event loop
+            # that is already closed, causing errors/warnings.
             _LOGGER.info("Received cancellation, stopping server gracefully...")
             await server.stop(NUMAFLOW_GRPC_SHUTDOWN_GRACE_PERIOD_SECONDS)
 

packages/pynumaflow/pynumaflow/sourcer/async_server.py

@@ -214,9 +214,12 @@ async def _watch_for_shutdown():
         try:
             await server.wait_for_termination()
         except asyncio.CancelledError:
-            # SIGTERM received — aiorun cancels all tasks. We must stop
-            # the gRPC server explicitly so its __del__ doesn't try to
-            # schedule a coroutine on the already-closed event loop.
+            # SIGTERM received — aiorun cancels all tasks. Unlike the UDF-error
+            # path (where _watch_for_shutdown calls server.stop()), this path
+            # must stop the gRPC server explicitly. Without this, the server
+            # object is never stopped and when it is garbage-collected, its
+            # __del__ tries to schedule a cleanup coroutine on an event loop
+            # that is already closed, causing errors/warnings.
             _LOGGER.info("Received cancellation, stopping server gracefully...")
             await server.stop(NUMAFLOW_GRPC_SHUTDOWN_GRACE_PERIOD_SECONDS)
 

packages/pynumaflow/pynumaflow/sourcetransformer/async_server.py

@@ -176,9 +176,12 @@ async def _watch_for_shutdown():
         try:
             await server.wait_for_termination()
         except asyncio.CancelledError:
-            # SIGTERM received — aiorun cancels all tasks. We must stop
-            # the gRPC server explicitly so its __del__ doesn't try to
-            # schedule a coroutine on the already-closed event loop.
+            # SIGTERM received — aiorun cancels all tasks. Unlike the UDF-error
+            # path (where _watch_for_shutdown calls server.stop()), this path
+            # must stop the gRPC server explicitly. Without this, the server
+            # object is never stopped and when it is garbage-collected, its
+            # __del__ tries to schedule a cleanup coroutine on an event loop
+            # that is already closed, causing errors/warnings.
             _LOGGER.info("Received cancellation, stopping server gracefully...")
             await server.stop(NUMAFLOW_GRPC_SHUTDOWN_GRACE_PERIOD_SECONDS)