tested all for graceful shutdown on kubectl delete

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

Author: BulkBeing

packages/pynumaflow/pynumaflow/mapper/multiproc_server.py

@@ -130,7 +130,7 @@ def start(self) -> None:
         server_info.metadata[MAP_MODE_KEY] = MapMode.UnaryMap
 
         # Start the multiproc server
-        start_multiproc_server(
+        has_error = start_multiproc_server(
             max_threads=self.max_threads,
             servicer=self.servicer,
             process_count=self._process_count,
@@ -141,6 +141,6 @@ def start(self) -> None:
             shutdown_event=self._shutdown_event,
         )
 
-        if self._shutdown_event.is_set():
+        if has_error:
             _LOGGER.critical("Server exiting due to worker error")
             sys.exit(1)

packages/pynumaflow/pynumaflow/shared/server.py

@@ -206,6 +206,10 @@ def _sigterm_handler(signum, frame):
     for worker in workers:
         worker.join()
 
+    # Return True if any worker exited with a non-zero code (i.e. a real error,
+    # not a clean SIGTERM shutdown).
+    return any(w.exitcode != 0 for w in workers)
+
 
 async def start_async_server(
     server_async: grpc.aio.Server,

packages/pynumaflow/pynumaflow/sourcetransformer/multiproc_server.py

@@ -149,7 +149,7 @@ def start(self):
         serv_info.minimum_numaflow_version = MINIMUM_NUMAFLOW_VERSION[
             ContainerType.Sourcetransformer
         ]
-        start_multiproc_server(
+        has_error = start_multiproc_server(
             max_threads=self.max_threads,
             servicer=self.servicer,
             process_count=self._process_count,
@@ -160,6 +160,6 @@ def start(self):
             shutdown_event=self._shutdown_event,
         )
 
-        if self._shutdown_event.is_set():
+        if has_error:
             _LOGGER.critical("Server exiting due to worker error")
             sys.exit(1)