core: Do not leak server state when application callbacks throw exceptions

core/src/main/java/io/grpc/util/interceptor/server/StatusRuntimeExceptionTransmitter.java → core/src/main/java/io/grpc/util/TransmitStatusRuntimeExceptionInterceptor.java

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package io.grpc.util.interceptor.server;
+package io.grpc.util;
 
 import com.google.common.util.concurrent.MoreExecutors;
 import com.google.common.util.concurrent.SettableFuture;
@@ -34,7 +34,8 @@
 
 /**
  * A class that intercepts uncaught exceptions of type {@link StatusRuntimeException} and handles
- * them by closing the {@link ServerCall}, and transmitting the exception's details to the client.
+ * them by closing the {@link ServerCall}, and transmitting the exception's status and metadata
+ * to the client.
  *
  * <p>Without this interceptor, gRPC will strip all details and close the {@link ServerCall} with
  * a generic {@link Status#UNKNOWN} code.
@@ -44,13 +45,12 @@
  * if all clients are trusted.
  */
 @ExperimentalApi("https://github.com/grpc/grpc-java/issues/2189")
-public final class StatusRuntimeExceptionTransmitter implements ServerInterceptor {
-  private StatusRuntimeExceptionTransmitter() {
-    // do not instantiate
+public final class TransmitStatusRuntimeExceptionInterceptor implements ServerInterceptor {
+  private TransmitStatusRuntimeExceptionInterceptor() {
   }
 
   public static ServerInterceptor instance() {
-    return new StatusRuntimeExceptionTransmitter();
+    return new TransmitStatusRuntimeExceptionInterceptor();
   }
 
   @Override
@@ -105,11 +105,7 @@ public void onReady() {
       }
 
       private void closeWithException(StatusRuntimeException t) {
-        Metadata metadata = Status.trailersFromThrowable(t);
-        if (metadata == null) {
-          metadata = new Metadata();
-        }
-        serverCall.close(t.getStatus(), metadata);
+        serverCall.close(t.getStatus(), t.getTrailers());
       }
     };
   }

core/src/test/java/io/grpc/util/interceptor/server/UtilServerInterceptorsTest.java

@@ -17,16 +17,12 @@
 package io.grpc.util.interceptor.server;
 
 import static com.google.common.collect.Iterables.getOnlyElement;
-import static org.mockito.Matchers.any;
 import static org.mockito.Matchers.same;
-import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 
 import io.grpc.Metadata;
 import io.grpc.MethodDescriptor;
-import io.grpc.MethodDescriptor.Marshaller;
-import io.grpc.MethodDescriptor.MethodType;
 import io.grpc.ServerCall;
 import io.grpc.ServerCallHandler;
 import io.grpc.ServerInterceptors;
@@ -36,12 +32,13 @@
 import io.grpc.Status;
 import io.grpc.StatusRuntimeException;
 import io.grpc.testing.NoopServerCall;
+import io.grpc.testing.TestMethodDescriptors;
+import io.grpc.util.TransmitStatusRuntimeExceptionInterceptor;
 import java.util.Arrays;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.JUnit4;
-import org.mockito.Mock;
 import org.mockito.Mockito;
 import org.mockito.MockitoAnnotations;
 
@@ -51,42 +48,27 @@
  */
 @RunWith(JUnit4.class)
 public class UtilServerInterceptorsTest {
-  @Mock
-  private Marshaller<String> requestMarshaller;
-
-  @Mock
-  private Marshaller<Integer> responseMarshaller;
+  private MethodDescriptor<String, Integer> flowMethod = TestMethodDescriptors.noopMethod();
+  private ServerCall<String, Integer> call = Mockito.spy(new NoopServerCall<String, Integer>());
+  private final Metadata headers = new Metadata();
 
-  @Mock
+  private ServerServiceDefinition serviceDefinition;
   private ServerCallHandler<String, Integer> handler;
-
-  @Mock
   private ServerCall.Listener<String> listener;
 
-  private MethodDescriptor<String, Integer> flowMethod;
-
-  private ServerCall<String, Integer> call = new NoopServerCall<String, Integer>();
-
-  private ServerServiceDefinition serviceDefinition;
-
-  private final Metadata headers = new Metadata();
-
   /** Set up for test. */
   @Before
   public void setUp() {
     MockitoAnnotations.initMocks(this);
-    flowMethod = MethodDescriptor.<String, Integer>newBuilder()
-        .setType(MethodType.UNKNOWN)
-        .setFullMethodName("basic/flow")
-        .setRequestMarshaller(requestMarshaller)
-        .setResponseMarshaller(responseMarshaller)
-        .build();
-
-    Mockito.when(handler.startCall(
-        Mockito.<ServerCall<String, Integer>>any(), Mockito.<Metadata>any()))
-        .thenReturn(listener);
-
-    serviceDefinition = ServerServiceDefinition.builder(new ServiceDescriptor("basic", flowMethod))
+    handler = new ServerCallHandler<String, Integer>() {
+      @Override
+      public ServerCall.Listener<String> startCall(
+          ServerCall<String, Integer> call, Metadata headers) {
+        return listener;
+      }
+    };
+    serviceDefinition = ServerServiceDefinition.builder(
+        new ServiceDescriptor("service_foo", flowMethod))
         .addMethod(flowMethod, handler).build();
   }
 
@@ -103,18 +85,38 @@ private static ServerMethodDefinition<String, Integer> getSoleMethod(
   public void statusRuntimeExceptionTransmitter() {
     final Status expectedStatus = Status.UNAVAILABLE;
     final Metadata expectedMetadata = new Metadata();
-    call = Mockito.spy(call);
     final StatusRuntimeException exception =
         new StatusRuntimeException(expectedStatus, expectedMetadata);
-    doThrow(exception).when(listener).onMessage(any(String.class));
-    doThrow(exception).when(listener).onCancel();
-    doThrow(exception).when(listener).onComplete();
-    doThrow(exception).when(listener).onHalfClose();
-    doThrow(exception).when(listener).onReady();
+    listener = new ServerCall.Listener<String>() {
+      @Override
+      public void onMessage(String message) {
+        throw exception;
+      }
+
+      @Override
+      public void onHalfClose() {
+        throw exception;
+      }
+
+      @Override
+      public void onCancel() {
+        throw exception;
+      }
+
+      @Override
+      public void onComplete() {
+        throw exception;
+      }
+
+      @Override
+      public void onReady() {
+        throw exception;
+      }
+    };
 
     ServerServiceDefinition intercepted = ServerInterceptors.intercept(
         serviceDefinition,
-        Arrays.asList(StatusRuntimeExceptionTransmitter.instance()));
+        Arrays.asList(TransmitStatusRuntimeExceptionInterceptor.instance()));
     // The interceptor should have handled the error by directly closing the ServerCall
     // and the exception should not propagate to the method's caller
     getSoleMethod(intercepted).getServerCallHandler().startCall(call, headers).onMessage("hello");