How to add global exception interceptor in gRPC server?

Question

In gRPC , how to add a global exception interceptor that intercepts any RuntimeException and propagate meaningful information to the client ?

for example , a divide method may throw ArithmeticException with / by zero message . In the server side , I may write :

@Override
public void divide(DivideRequest request, StreamObserver<DivideResponse> responseObserver) {
  int dom = request.getDenominator();
  int num = request.getNumerator();

  double result = num / dom;
  responseObserver.onNext(DivideResponse.newBuilder().setValue(result).build());
  responseObserver.onCompleted();
}

If the client passes denominator = 0 , it will get :

Exception in thread "main" io.grpc.StatusRuntimeException: UNKNOWN

And the server outputs

Exception while executing runnable io.grpc.internal.ServerImpl$JumpToApplicationThreadServerStreamListener$2@62e95ade
java.lang.ArithmeticException: / by zero

The client doesn't know what's going on.

If I want to pass / by zero message to client , I have to modify server to : (as described in this question )

  try {
    double result = num / dom;
    responseObserver.onNext(DivideResponse.newBuilder().setValue(result).build());
    responseObserver.onCompleted();
  } catch (Exception e) {
    logger.error("onError : {}" , e.getMessage());
    responseObserver.onError(new StatusRuntimeException(Status.INTERNAL.withDescription(e.getMessage())));
  }

And if client sends denominator = 0 , it will get :

Exception in thread "main" io.grpc.StatusRuntimeException: INTERNAL: / by zero

Good , / by zero is passed to client.

But the problem is , in a truly enterprise environment, there will be a lot of RuntimeExceptions , and if I want to pass these exception's messages to client , I will have to try catch each method , which is very cumbersome.

Is there any global interceptor that intercepts every method , catching RuntimeException and trigger onError and propagate the error message to client ? So that I don't have to deal with RuntimeExceptions in my server code .

Thanks a lot !

Note :

<grpc.version>1.0.1</grpc.version>
com.google.protobuf:proton:3.1.0
io.grpc:protoc-gen-grpc-java:1.0.1

Custommade

Answer 1

Below code will catch all runtime exceptions, Also refer the link https://github.com/grpc/grpc-java/issues/1552

public class GlobalGrpcExceptionHandler implements ServerInterceptor {

   @Override
   public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(ServerCall<ReqT, RespT> call,
         Metadata requestHeaders, ServerCallHandler<ReqT, RespT> next) {
      ServerCall.Listener<ReqT> delegate = next.startCall(call, requestHeaders);
      return new SimpleForwardingServerCallListener<ReqT>(delegate) {
         @Override
         public void onHalfClose() {
            try {
               super.onHalfClose();
            } catch (Exception e) {
               call.close(Status.INTERNAL
                .withCause (e)
                .withDescription("error message"), new Metadata());
            }
         }
      };
   }
}

Certify

Answer 2

If you want to catch exceptions in all gRPC endpoints (including the ones processing client streams) and interceptors, you probably want something similar to the following:

import io.grpc.ForwardingServerCallListener;
import io.grpc.Metadata;
import io.grpc.ServerCall;
import io.grpc.ServerCallHandler;
import io.grpc.ServerInterceptor;
import io.grpc.Status;

public class GlobalGrpcExceptionHandler implements ServerInterceptor {

    @Override
    public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(ServerCall<ReqT, RespT> serverCall,
                                                                 Metadata requestHeaders,
                                                                 ServerCallHandler<ReqT, RespT> serverCallHandler) {
        try {
            ServerCall.Listener<ReqT> delegate = serverCallHandler.startCall(serverCall, requestHeaders);
            return new ForwardingServerCallListener.SimpleForwardingServerCallListener<ReqT>(delegate) {
                @Override
                public void onMessage(ReqT message) {
                    try {
                        super.onMessage(message); // Here onNext is called (in case of client streams)
                    } catch (Throwable e) {
                        handleEndpointException(e, serverCall);
                    }
                }

                @Override
                public void onHalfClose() {
                    try {
                        super.onHalfClose(); // Here onCompleted is called (in case of client streams)
                    } catch (Throwable e) {
                        handleEndpointException(e, serverCall);
                    }
                }
            };
        } catch (Throwable t) {
            return handleInterceptorException(t, serverCall);
        }
    }

    private <ReqT, RespT> void handleEndpointException(Throwable t, ServerCall<ReqT, RespT> serverCall) {
        serverCall.close(Status.INTERNAL
                         .withCause(t)
                         .withDescription("An exception occurred in the endpoint implementation"), new Metadata());
    }

    private <ReqT, RespT> ServerCall.Listener<ReqT> handleInterceptorException(Throwable t, ServerCall<ReqT, RespT> serverCall) {
        serverCall.close(Status.INTERNAL
                         .withCause(t)
                         .withDescription("An exception occurred in a **subsequent** interceptor"), new Metadata());

        return new ServerCall.Listener<ReqT>() {
            // no-op
        };
    }
}

DISCLAIMER: I have gathered this by inspecting the implementation, I have not read it in the documentation and I'm not sure if it will change. For reference, I'm referring to io.grpc version 1.30.

Plasticine

Answer 3

TransmitStatusRuntimeExceptionInterceptor is very similar to what you want, except that it only catches StatusRuntimeException. You can fork it and make it catch all exceptions.

To install an interceptor for all services on a server, you can use ServerBuilder.intercept(), which was added in gRPC 1.5.0

Gregg

Answer 4

In Kotlin, you have to structure your ServerInterceptor differently. I was using grpc-kotlin in Micronaut and the exceptions never appeared in the SimpleForwardingServerCallListener onHalfClose or other handlers.

Broida

Answer 5

If you can convert your gRPC server application to spring boot using yidongnan/grpc-spring-boot-starter, then you can write @GrpcAdvice, similar to Spring Boot @ControllerAdvice as

    @GrpcAdvice
    public class ExceptionHandler {
    
      @GrpcExceptionHandler(ValidationErrorException.class)
      public StatusRuntimeException handleValidationError(ValidationErrorException cause) {
    
         Status.INVALID_ARGUMENT.withDescription("Invalid Argument")
            .asRuntimeException()
      }
    }

Wheeler

Answer 6

Have you read the grpc java examples for interceptor?

So in my case, we use code and message as standard for defining what kind of error the server sent onto client.

Examples: server send response like

{
  code: 409,
  message: 'Id xxx aldready exist'
}

So in the client you can setup client interceptor for getting that code and response with Reflection. Fyi we use Lognet Spring Boot starter for grpc as the server, and Spring boot for the client.

Sura

Answer 7

public class GrpcExceptionHandler implements ServerInterceptor {

private final Logger logger = LoggerFactory.getLogger (GrpcExceptionHandler.class);

@Override
public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall (ServerCall<ReqT, RespT> call,
                                                              Metadata headers,
                                                              ServerCallHandler<ReqT, RespT> next) {
    logger.info ("GRPC call at: {}", Instant.now ());
    ServerCall.Listener<ReqT> listener;

    try {
        listener = next.startCall (call, headers);
    } catch (Throwable ex) {
        logger.error ("Uncaught exception from grpc service");
        call.close (Status.INTERNAL
                .withCause (ex)
                .withDescription ("Uncaught exception from grpc service"), null);
        return new ServerCall.Listener<ReqT>() {};
    }

    return listener;
}

}

Sample interceptor above.

You need to bootstrap it, of course, before expecting anything from it;

serverBuilder.addService (ServerInterceptors.intercept (bindableService, interceptor));

UPDATE

public interface ServerCallHandler<RequestT, ResponseT> {
  /**
   * Produce a non-{@code null} listener for the incoming call. Implementations are free to call
   * methods on {@code call} before this method has returned.
   *
   * <p>If the implementation throws an exception, {@code call} will be closed with an error.
   * Implementations must not throw an exception if they started processing that may use {@code
   * call} on another thread.
   *
   * @param call object for responding to the remote client.
   * @return listener for processing incoming request messages for {@code call}
   */
  ServerCall.Listener<RequestT> startCall(
      ServerCall<RequestT, ResponseT> call,
      Metadata headers);
}

Sadly, different thread context means no exception handling scope, so my answer is not the solution you are looking..

Tl

Sorry , it does not work. This line logger.error ("Uncaught exception from grpc service"); is not reached ! I feel strange too. – Custommade

Well, interception do occurs but as the documentation states, next.startCall(call, headers) immediately returns and get executed in another thread and eventually we loose the stack scope for the exception. Sadly I don't know if any workaround is possible at the moment. – Tl

Answer 8

On Kotlin adding the try/catch on the listener's methods does not work, for some reason the exceptions are swallowed.

Following the link that @markficket posted, I've implemented a solution creating an implementation of a SimpleForwardingServerCall.

class ErrorHandlerServerInterceptor : ServerInterceptor {

    private inner class StatusExceptionHandlingServerCall<ReqT, RespT>(delegate: ServerCall<ReqT, RespT>) 
        : ForwardingServerCall.SimpleForwardingServerCall<ReqT, RespT>(delegate) {

        override fun close(status: Status, trailers: Metadata) {
            status.run {
                when {
                    isOk -> status
                    cause is MyException -> myExceptionHandler(cause as MyException)
                    else -> defaultExceptionHandler(cause)
                }
            }
                .let { super.close(it, trailers) }
        }

        private fun myExceptionHandler(cause: MyException): Status = cause.run { ... }

        private fun defaultExceptionHandler(cause: Throwable?): Status = cause?.run { ... }

    }

    override fun <ReqT : Any, RespT : Any> interceptCall(
        call: ServerCall<ReqT, RespT>,
        metadata: Metadata,
        next: ServerCallHandler<ReqT, RespT>
    ): ServerCall.Listener<ReqT> =
        next.startCall(StatusExceptionHandlingServerCall(call), metadata)

}

Then of course you need to add the interceptor on the server creation

ServerBuilder
  .forPort(port)
  .executor(Dispatchers.IO.asExecutor())
  .addService(service)
  .intercept(ErrorHandlerServerInterceptor())
  .build()

And then you can simply throw exceptions on your gRPC methods

override suspend fun someGrpcCall(request: Request): Response {
  ... code ...
  throw NotFoundMyException("Cannot find entity")
}

Mckamey

Answer 9

I have used AOP to deal with rpc errors globally, and I find it convenient. I use the AOP in guice, and the way to use it in spring should be similar

1. define a method interceptor

```

public class ServerExceptionInterceptor implements MethodInterceptor {
    final static Logger logger = LoggerFactory.getLogger(ServerExceptionInterceptor.class);

    public Object invoke(MethodInvocation invocation) throws Throwable {
        try {
            return  invocation.proceed();
        } catch (Exception ex) {
            String stackTrace = Throwables.getStackTraceAsString(ex);
            logger.error("##grpc server side error, {}", stackTrace);
            Object[] args = invocation.getArguments();
            StreamObserver<?> responseObserver = (StreamObserver<?>)args[1];
            responseObserver.onError(Status.INTERNAL
                    .withDescription(stackTrace)
                    .withCause(ex)
                    .asRuntimeException());

            return null;
        }
    }

    @Target({ElementType.METHOD, ElementType.TYPE}) @Retention(RUNTIME)
    public @interface WrapError {
        String value() default "";
    }

}

```

2. add @WrapError to all rpc method @Override @WrapError public void sayHello(HelloRequest req, StreamObserver<HelloReply> responseObserver) { HelloReply reply = HelloReply.newBuilder().setMessage("Hello " + req.getName()).build(); logger.info("#rpc server, sayHello, planId: {}", req.getName()); if(true) throw new RuntimeException("testing-rpc-error"); //simulate an exception responseObserver.onNext(reply); responseObserver.onCompleted(); }

   3. bind the interceptor in guice module

ServerExceptionInterceptor interceptor = new ServerExceptionInterceptor(); requestInjection(interceptor); bindInterceptor(Matchers.any(), Matchers.annotatedWith(WrapError.class), interceptor);

4.testing

Collywobbles