Trying to find cases if the ServerCall listener callbacks for onComplete don't get called if there is an application error.

Author: kannanjgithub

examples/build.gradle

@@ -45,7 +45,7 @@ dependencies {
 protobuf {
     protoc { artifact = "com.google.protobuf:protoc:${protocVersion}" }
     plugins {
-        grpc { artifact = "io.grpc:protoc-gen-grpc-java:${grpcVersion}" }
+        grpc { artifact = "io.grpc:protoc-gen-grpc-java:1.73.0" }
     }
     generateProtoTasks {
         all()*.plugins { grpc {} }

examples/example-oauth/build.gradle

@@ -44,7 +44,7 @@ dependencies {
 protobuf {
     protoc { artifact = "com.google.protobuf:protoc:${protocVersion}" }
     plugins {
-        grpc { artifact = "io.grpc:protoc-gen-grpc-java:${grpcVersion}" }
+        grpc { artifact = "io.grpc:protoc-gen-grpc-java:1.73.0" }
     }
     generateProtoTasks {
         all()*.plugins { grpc {} }

examples/example-oauth/src/main/java/io/grpc/examples/oauth/AuthServer.java

@@ -93,12 +93,7 @@ public static void main(String[] args) throws IOException, InterruptedException
   static class GreeterImpl extends GreeterGrpc.GreeterImplBase {
     @Override
     public void sayHello(HelloRequest req, StreamObserver<HelloReply> responseObserver) {
-      // get client id added to context by interceptor
-      String clientId = Constant.CLIENT_ID_CONTEXT_KEY.get();
-      logger.info("Processing request from " + clientId);
-      HelloReply reply = HelloReply.newBuilder().setMessage("Hello, " + req.getName()).build();
-      responseObserver.onNext(reply);
-      responseObserver.onCompleted();
+      throw new ArithmeticException("arith");
     }
   }
 }

examples/src/main/java/io/grpc/examples/header/CustomHeaderServer.java

@@ -88,9 +88,7 @@ private static class GreeterImpl extends GreeterGrpc.GreeterImplBase {
 
     @Override
     public void sayHello(HelloRequest req, StreamObserver<HelloReply> responseObserver) {
-      HelloReply reply = HelloReply.newBuilder().setMessage("Hello " + req.getName()).build();
-      responseObserver.onNext(reply);
-      responseObserver.onCompleted();
+      throw new ArithmeticException("arith");
     }
   }
 }

examples/src/main/java/io/grpc/examples/header/HeaderServerInterceptor.java

@@ -17,6 +17,7 @@
 package io.grpc.examples.header;
 
 import com.google.common.annotations.VisibleForTesting;
+import io.grpc.ForwardingServerCall;
 import io.grpc.ForwardingServerCall.SimpleForwardingServerCall;
 import io.grpc.Metadata;
 import io.grpc.ServerCall;
@@ -42,12 +43,39 @@ public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(
       final Metadata requestHeaders,
       ServerCallHandler<ReqT, RespT> next) {
     logger.info("header received from client:" + requestHeaders);
-    return next.startCall(new SimpleForwardingServerCall<ReqT, RespT>(call) {
+    ServerCall.Listener<ReqT> serverCallListener = next.startCall(new SimpleForwardingServerCall<ReqT, RespT>(call) {
       @Override
       public void sendHeaders(Metadata responseHeaders) {
         responseHeaders.put(CUSTOM_HEADER_KEY, "customRespondValue");
         super.sendHeaders(responseHeaders);
       }
     }, requestHeaders);
+
+    return new ServerCall.Listener<ReqT>() {
+      @Override
+      public void onMessage(ReqT message) {
+        serverCallListener.onMessage(message);
+      }
+
+      @Override
+      public void onComplete() {
+        serverCallListener.onComplete();
+      }
+
+      @Override
+      public void onCancel() {
+        serverCallListener.onCancel();
+      }
+
+      @Override
+      public void onHalfClose() {
+        serverCallListener.onHalfClose();
+      }
+
+      @Override
+      public void onReady() {
+        serverCallListener.onReady();
+      }
+    };
   }
 }

examples/src/main/java/io/grpc/examples/manualflowcontrol/ManualFlowControlServer.java

@@ -18,7 +18,9 @@
 
 import io.grpc.Server;
 import io.grpc.ServerBuilder;
+import io.grpc.ServerInterceptors;
 import io.grpc.Status;
+import io.grpc.examples.header.HeaderServerInterceptor;
 import io.grpc.stub.ServerCallStreamObserver;
 import io.grpc.stub.StreamObserver;
 
@@ -76,15 +78,21 @@ public void run() {
           @Override
           public void onNext(HelloRequest request) {
             // Process the request and send a response or an error.
-            try {
+
               // Accept and enqueue the request.
               String name = request.getName();
               logger.info("--> " + name);
 
               // Simulate server "work"
               int sleepMillis = ++cnt % 20 == 0 ? 2000 : 100;
-              Thread.sleep(sleepMillis);
-
+              try {
+                Thread.sleep(sleepMillis);
+              } catch (InterruptedException e) {
+                throw new RuntimeException(e);
+              }
+              if (cnt > 1) {
+                throw new ArithmeticException("arith");
+              }
               // Send a response.
               String message = "Hello " + name;
               logger.info("<-- " + message);
@@ -106,11 +114,6 @@ public void onNext(HelloRequest request) {
                 // If not, note that back-pressure has begun.
                 onReadyHandler.wasReady = false;
               }
-            } catch (Throwable throwable) {
-              throwable.printStackTrace();
-              responseObserver.onError(
-                  Status.UNKNOWN.withDescription("Error handling request").withCause(throwable).asException());
-            }
           }
 
           @Override
@@ -132,7 +135,7 @@ public void onCompleted() {
 
     final Server server = ServerBuilder
         .forPort(50051)
-        .addService(svc)
+        .addService(ServerInterceptors.intercept(svc, new HeaderServerInterceptor()))
         .build()
         .start();