chore: graceful shutdown of sink server

src/main/java/io/numaproj/numaflow/sinker/Server.java

@@ -7,6 +7,7 @@
 import io.numaproj.numaflow.shared.GrpcServerUtils;
 import lombok.extern.slf4j.Slf4j;
 
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
 
 /**
@@ -17,6 +18,7 @@ public class Server {
 
     private final GRPCConfig grpcConfig;
     private final Service service;
+    public final CompletableFuture<Void> shutdownSignal;
     private final ServerInfoAccessor serverInfoAccessor = new ServerInfoAccessorImpl(new ObjectMapper());
     private io.grpc.Server server;
 
@@ -36,7 +38,8 @@ public Server(Sinker sinker) {
      * @param sinker sink to process the message
      */
     public Server(Sinker sinker, GRPCConfig grpcConfig) {
-        this.service = new Service(sinker);
+        this.shutdownSignal = new CompletableFuture<>();
+        this.service = new Service(sinker, this.shutdownSignal);
         this.grpcConfig = grpcConfig;
     }
 
@@ -76,15 +79,36 @@ public void start() throws Exception {
 
         // register shutdown hook
         Runtime.getRuntime().addShutdownHook(new Thread(() -> {
+            if (server.isTerminated()) {
+                return;
+            }
             // Use stderr here since the logger may have been reset by its JVM shutdown hook.
-            System.err.println("*** shutting down gRPC server since JVM is shutting down");
+            System.err.println("*** shutting down sink gRPC server since JVM is shutting down");
             try {
                 Server.this.stop();
             } catch (InterruptedException e) {
                 Thread.interrupted();
                 e.printStackTrace(System.err);
             }
         }));
+
+        // if there are any exceptions, shutdown the server gracefully.
+        shutdownSignal.whenCompleteAsync((v, e) -> {
+            if (server.isTerminated()) {
+                return;
+            }
+
+            if (e != null) {
+                System.err.println("*** shutting down sink gRPC server because of an exception - " + e.getMessage());
+                try {
+                    Server.this.stop();
+                } catch (InterruptedException ex) {
+                    Thread.interrupted();
+                    ex.printStackTrace(System.err);
+                }
+            }
+            System.exit(0);
+        });
     }
 
     /**

src/main/java/io/numaproj/numaflow/sinker/Service.java

@@ -1,13 +1,15 @@
 package io.numaproj.numaflow.sinker;
 
 import com.google.protobuf.Empty;
+import io.grpc.Status;
 import io.grpc.stub.StreamObserver;
 import io.numaproj.numaflow.sink.v1.SinkGrpc;
 import io.numaproj.numaflow.sink.v1.SinkOuterClass;
 import lombok.extern.slf4j.Slf4j;
 
 import java.time.Instant;
 import java.util.List;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -31,9 +33,11 @@ class Service extends SinkGrpc.SinkImplBase {
 
 
     private final Sinker sinker;
+    private final CompletableFuture<Void> shutdownSignal;
 
-    public Service(Sinker sinker) {
+    public Service(Sinker sinker, CompletableFuture<Void> shutdownSignal) {
         this.sinker = sinker;
+        this.shutdownSignal = shutdownSignal;
     }
 
     /**
@@ -52,7 +56,7 @@ public StreamObserver<SinkOuterClass.SinkRequest> sinkFn(StreamObserver<SinkOute
         Future<ResponseList> result = sinkTaskExecutor.submit(() -> this.sinker.processMessages(
                 datumStream));
 
-        return new StreamObserver<SinkOuterClass.SinkRequest>() {
+        return new StreamObserver<>() {
             @Override
             public void onNext(SinkOuterClass.SinkRequest d) {
                 try {
@@ -83,7 +87,8 @@ public void onCompleted() {
 
                 } catch (InterruptedException | ExecutionException e) {
                     e.printStackTrace();
-                    onError(e);
+                    shutdownSignal.completeExceptionally(e);
+                    responseObserver.onError(Status.UNKNOWN.withDescription(e.getMessage()).asException());
                 }
                 responseObserver.onNext(response);
                 responseObserver.onCompleted();