diff --git a/core/src/main/java/io/grpc/internal/ServerImpl.java b/core/src/main/java/io/grpc/internal/ServerImpl.java index dc0709e1fb8..9af2f8a2cf9 100644 --- a/core/src/main/java/io/grpc/internal/ServerImpl.java +++ b/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 diff --git a/core/src/test/java/io/grpc/internal/ServerImplTest.java b/core/src/test/java/io/grpc/internal/ServerImplTest.java index 0f18efe078c..1027fe53681 100644 --- a/core/src/test/java/io/grpc/internal/ServerImplTest.java +++ b/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();