11246 :: Unexpected error when server expands a compressed message to learn it is too large

core/src/main/java/io/grpc/internal/ServerImpl.java

@@ -56,6 +56,7 @@
 import io.grpc.ServerServiceDefinition;
 import io.grpc.ServerTransportFilter;
 import io.grpc.Status;
+import io.grpc.StatusRuntimeException;
 import io.perfmark.Link;
 import io.perfmark.PerfMark;
 import io.perfmark.Tag;
@@ -811,7 +812,13 @@ void setListener(ServerStreamListener listener) {
     private void internalClose(Throwable t) {
       // TODO(ejona86): this is not thread-safe :)
       String description = "Application error processing RPC";
-      stream.close(Status.UNKNOWN.withDescription(description).withCause(t), new Metadata());
+      Status statusToPropagate = Status.UNKNOWN.withDescription(description).withCause(t);
+      if (t instanceof StatusRuntimeException) {
+        if (((StatusRuntimeException) t).getStatus().getCode() == Status.Code.RESOURCE_EXHAUSTED) {
+          statusToPropagate = ((StatusRuntimeException) t).getStatus().withCause(t);
+        }
+      }
+      stream.close(statusToPropagate, new Metadata());
     }
 
     @Override

core/src/test/java/io/grpc/internal/ServerImplTest.java

@@ -77,6 +77,7 @@
 import io.grpc.ServiceDescriptor;
 import io.grpc.Status;
 import io.grpc.Status.Code;
+import io.grpc.StatusRuntimeException;
 import io.grpc.StringMarshaller;
 import io.grpc.internal.ServerImpl.JumpToApplicationThreadServerStreamListener;
 import io.grpc.internal.ServerImplBuilder.ClientTransportServersBuilder;
@@ -1542,6 +1543,99 @@ public void channelz_transport_membershp() throws Exception {
     assertTrue(after.end);
   }
 
+  @Test
+  public void testInternalClose_nonProtocolStatusRuntimeExceptionBecomesUnknown() {
+    JumpToApplicationThreadServerStreamListener listener
+        = new JumpToApplicationThreadServerStreamListener(
+            executor.getScheduledExecutorService(),
+            executor.getScheduledExecutorService(),
+            stream,
+            Context.ROOT.withCancellation(),
+            PerfMark.createTag());
+    ServerStreamListener mockListener = mock(ServerStreamListener.class);
+    listener.setListener(mockListener);
+
+    StatusRuntimeException statusRuntimeException
+        = new StatusRuntimeException(Status.PERMISSION_DENIED.withDescription("denied"));
+    doThrow(statusRuntimeException).when(mockListener).onReady();
+    listener.onReady();
+    try {
+      executor.runDueTasks();
+      fail("Expected exception");
+    } catch (RuntimeException t) {
+      assertSame(statusRuntimeException, t);
+      ensureServerStateNotLeaked();
+    }
+    verify(stream).close(statusCaptor.capture(), metadataCaptor.capture());
+    Status status = statusCaptor.getValue();
+    assertEquals(Code.UNKNOWN, status.getCode());
+    assertEquals("Application error processing RPC", status.getDescription());
+    assertEquals(statusRuntimeException, status.getCause());
+    assertTrue(metadataCaptor.getValue().keys().isEmpty());
+  }
+
+  @Test
+  public void testInternalClose_otherExceptionBecomesUnknown() {
+    JumpToApplicationThreadServerStreamListener listener
+        = new JumpToApplicationThreadServerStreamListener(
+            executor.getScheduledExecutorService(),
+            executor.getScheduledExecutorService(),
+            stream,
+            Context.ROOT.withCancellation(),
+            PerfMark.createTag());
+    ServerStreamListener mockListener = mock(ServerStreamListener.class);
+    listener.setListener(mockListener);
+
+    RuntimeException expectedT = new RuntimeException();
+    doThrow(expectedT).when(mockListener)
+        .messagesAvailable(any(StreamListener.MessageProducer.class));
+    listener.messagesAvailable(mock(StreamListener.MessageProducer.class));
+    try {
+      executor.runDueTasks();
+      fail("Expected exception");
+    } catch (RuntimeException t) {
+      assertSame(expectedT, t);
+      ensureServerStateNotLeaked();
+    }
+    verify(stream).close(statusCaptor.capture(), metadataCaptor.capture());
+    Status status = statusCaptor.getValue();
+    assertEquals(Code.UNKNOWN, status.getCode());
+    assertEquals("Application error processing RPC", status.getDescription());
+    assertEquals(expectedT, status.getCause());
+    assertTrue(metadataCaptor.getValue().keys().isEmpty());
+  }
+
+  @Test
+  public void testInternalClose_propagatesResourceExhausted() {
+    JumpToApplicationThreadServerStreamListener listener
+        = new JumpToApplicationThreadServerStreamListener(
+            executor.getScheduledExecutorService(),
+            executor.getScheduledExecutorService(),
+            stream,
+            Context.ROOT.withCancellation(),
+            PerfMark.createTag());
+    ServerStreamListener mockListener = mock(ServerStreamListener.class);
+    listener.setListener(mockListener);
+
+    StatusRuntimeException statusRuntimeException
+            = new StatusRuntimeException(Status.RESOURCE_EXHAUSTED.withDescription("exhausted"));
+    doThrow(statusRuntimeException).when(mockListener)
+        .messagesAvailable(any(StreamListener.MessageProducer.class));
+    listener.messagesAvailable(mock(StreamListener.MessageProducer.class));
+    try {
+      executor.runDueTasks();
+      fail("Expected exception");
+    } catch (RuntimeException t) {
+      assertSame(statusRuntimeException, t);
+    }
+    verify(stream).close(statusCaptor.capture(), metadataCaptor.capture());
+    Status status = statusCaptor.getValue();
+    assertEquals(Status.Code.RESOURCE_EXHAUSTED, status.getCode());
+    assertEquals("exhausted", status.getDescription());
+    assertEquals(statusRuntimeException, status.getCause());
+    assertTrue(metadataCaptor.getValue().keys().isEmpty());
+  }
+
   private void createAndStartServer() throws IOException {
     createServer();
     server.start();