Revert "core: Report marshaller error for uncompressed size too large back to the client "

This reverts commit a40d549.

A user has noticed it caused test failures. cl/828098798

```
java.lang.AssertionError: Failed executing read operation
	at io.grpc.internal.CompositeReadableBuffer.execute(CompositeReadableBuffer.java:328)
	at io.grpc.internal.CompositeReadableBuffer.executeNoThrow(CompositeReadableBuffer.java:336)
	at io.grpc.internal.CompositeReadableBuffer.readBytes(CompositeReadableBuffer.java:151)
	at io.grpc.internal.ReadableBuffers$BufferInputStream.read(ReadableBuffers.java:377)
	at io.grpc.protobuf.lite.ProtoLiteUtils$MessageMarshaller.parse(ProtoLiteUtils.java:205)
	at io.grpc.protobuf.lite.ProtoLiteUtils$MessageMarshaller.parse(ProtoLiteUtils.java:133)
	at io.grpc.MethodDescriptor.parseRequest(MethodDescriptor.java:307)
	at io.grpc.ServerInterceptors$2$2.onMessage(ServerInterceptors.java:324)
```

Author: ejona86

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

@@ -327,20 +327,16 @@ private void messagesAvailableInternal(final MessageProducer producer) {
         return;
       }
 
+      InputStream message;
       try {
-        InputStream message;
         while ((message = producer.next()) != null) {
-          ReqT parsedMessage;
-          try (InputStream ignored = message) {
-            parsedMessage = call.method.parseRequest(message);
-          } catch (StatusRuntimeException e) {
+          try {
+            listener.onMessage(call.method.parseRequest(message));
+          } catch (Throwable t) {
             GrpcUtil.closeQuietly(message);
-            GrpcUtil.closeQuietly(producer);
-            call.cancelled = true;
-            call.close(e.getStatus(), new Metadata());
-            return;
+            throw t;
           }
-          listener.onMessage(parsedMessage);
+          message.close();
         }
       } catch (Throwable t) {
         GrpcUtil.closeQuietly(producer);

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

@@ -48,11 +48,9 @@
 import io.grpc.SecurityLevel;
 import io.grpc.ServerCall;
 import io.grpc.Status;
-import io.grpc.StatusRuntimeException;
 import io.grpc.internal.ServerCallImpl.ServerStreamListenerImpl;
 import io.perfmark.PerfMark;
 import java.io.ByteArrayInputStream;
-import java.io.IOException;
 import java.io.InputStream;
 import java.io.InputStreamReader;
 import org.junit.Before;
@@ -71,8 +69,6 @@ public class ServerCallImplTest {
 
   @Mock private ServerStream stream;
   @Mock private ServerCall.Listener<Long> callListener;
-  @Mock private StreamListener.MessageProducer messageProducer;
-  @Mock private InputStream message;
 
   private final CallTracer serverCallTracer = CallTracer.getDefaultFactory().create();
   private ServerCallImpl<Long, Long> call;
@@ -497,43 +493,6 @@ public void streamListener_unexpectedRuntimeException() {
     assertThat(e).hasMessageThat().isEqualTo("unexpected exception");
   }
 
-  @Test
-  public void streamListener_statusRuntimeException() throws IOException {
-    MethodDescriptor<Long, Long> failingParseMethod = MethodDescriptor.<Long, Long>newBuilder()
-        .setType(MethodType.UNARY)
-        .setFullMethodName("service/method")
-        .setRequestMarshaller(new LongMarshaller() {
-            @Override
-            public Long parse(InputStream stream) {
-              throw new StatusRuntimeException(Status.RESOURCE_EXHAUSTED
-                  .withDescription("Decompressed gRPC message exceeds maximum size"));
-            }
-        })
-        .setResponseMarshaller(new LongMarshaller())
-        .build();
-
-    call =  new ServerCallImpl<>(stream, failingParseMethod, requestHeaders, context,
-            DecompressorRegistry.getDefaultInstance(), CompressorRegistry.getDefaultInstance(),
-            serverCallTracer, PerfMark.createTag());
-
-    ServerStreamListenerImpl<Long> streamListener =
-            new ServerCallImpl.ServerStreamListenerImpl<>(call, callListener, context);
-
-    when(messageProducer.next()).thenReturn(message, (InputStream) null);
-    streamListener.messagesAvailable(messageProducer);
-    ArgumentCaptor<Status> statusCaptor = ArgumentCaptor.forClass(Status.class);
-    ArgumentCaptor<Metadata> metadataCaptor = ArgumentCaptor.forClass(Metadata.class);
-
-    verify(stream).close(statusCaptor.capture(), metadataCaptor.capture());
-    Status status = statusCaptor.getValue();
-    assertEquals(Status.RESOURCE_EXHAUSTED.getCode(), status.getCode());
-    assertEquals("Decompressed gRPC message exceeds maximum size", status.getDescription());
-
-    streamListener.halfClosed();
-    verify(callListener, never()).onHalfClose();
-    verify(callListener, never()).onMessage(any());
-  }
-
   private static class LongMarshaller implements Marshaller<Long> {
     @Override
     public InputStream stream(Long value) {

interop-testing/src/main/java/io/grpc/testing/integration/AbstractInteropTest.java

@@ -2024,7 +2024,7 @@ private void assertPayload(Payload expected, Payload actual) {
     }
   }
 
-  protected static void assertCodeEquals(Status.Code expected, Status actual) {
+  private static void assertCodeEquals(Status.Code expected, Status actual) {
     assertWithMessage("Unexpected status: %s", actual).that(actual.getCode()).isEqualTo(expected);
   }
 

interop-testing/src/test/java/io/grpc/testing/integration/TransportCompressionTest.java

@@ -17,7 +17,6 @@
 package io.grpc.testing.integration;
 
 import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertThrows;
 import static org.junit.Assert.assertTrue;
 
 import com.google.protobuf.ByteString;
@@ -38,8 +37,6 @@
 import io.grpc.ServerCall.Listener;
 import io.grpc.ServerCallHandler;
 import io.grpc.ServerInterceptor;
-import io.grpc.Status.Code;
-import io.grpc.StatusRuntimeException;
 import io.grpc.internal.GrpcUtil;
 import io.grpc.netty.InternalNettyChannelBuilder;
 import io.grpc.netty.InternalNettyServerBuilder;
@@ -56,9 +53,7 @@
 import java.io.OutputStream;
 import org.junit.Before;
 import org.junit.BeforeClass;
-import org.junit.Rule;
 import org.junit.Test;
-import org.junit.rules.TestName;
 import org.junit.runner.RunWith;
 import org.junit.runners.JUnit4;
 
@@ -89,16 +84,10 @@ public static void registerCompressors() {
     compressors.register(Codec.Identity.NONE);
   }
 
-  @Rule
-  public final TestName currentTest = new TestName();
-
   @Override
   protected ServerBuilder<?> getServerBuilder() {
     NettyServerBuilder builder = NettyServerBuilder.forPort(0, InsecureServerCredentials.create())
-        .maxInboundMessageSize(
-            DECOMPRESSED_MESSAGE_TOO_LONG_METHOD_NAME.equals(currentTest.getMethodName())
-                ? 1000
-                : AbstractInteropTest.MAX_MESSAGE_SIZE)
+        .maxInboundMessageSize(AbstractInteropTest.MAX_MESSAGE_SIZE)
         .compressorRegistry(compressors)
         .decompressorRegistry(decompressors)
         .intercept(new ServerInterceptor() {
@@ -137,22 +126,6 @@ public void compresses() {
     assertTrue(FZIPPER.anyWritten);
   }
 
-  private static final String DECOMPRESSED_MESSAGE_TOO_LONG_METHOD_NAME =
-      "decompressedMessageTooLong";
-
-  @Test
-  public void decompressedMessageTooLong() {
-    assertEquals(DECOMPRESSED_MESSAGE_TOO_LONG_METHOD_NAME, currentTest.getMethodName());
-    final SimpleRequest bigRequest = SimpleRequest.newBuilder()
-        .setPayload(Payload.newBuilder().setBody(ByteString.copyFrom(new byte[10_000])))
-        .build();
-    StatusRuntimeException e = assertThrows(StatusRuntimeException.class,
-        () -> blockingStub.withCompression("gzip").unaryCall(bigRequest));
-    assertCodeEquals(Code.RESOURCE_EXHAUSTED, e.getStatus());
-    assertEquals("Decompressed gRPC message exceeds maximum size 1000",
-        e.getStatus().getDescription());
-  }
-
   @Override
   protected NettyChannelBuilder createChannelBuilder() {
     NettyChannelBuilder builder = NettyChannelBuilder.forAddress(getListenAddress())