move marker cause

Author: panchenko

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

@@ -521,7 +521,6 @@ private void verifySize() {
       if (count > maxMessageSize) {
         throw Status.RESOURCE_EXHAUSTED
             .withDescription("Decompressed gRPC message exceeds maximum size " + maxMessageSize)
-            .withCause(new CloseWithHeadersMarker())
             .asRuntimeException();
       }
     }

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

@@ -279,6 +279,12 @@ private void handleInternalError(Throwable internalError) {
     serverCallTracer.reportCallEnded(false); // error so always false
   }
 
+  private void handleParseError(StatusRuntimeException parseError) {
+    log.log(Level.WARNING, "Cancelling the stream because of parse error", parseError);
+    stream.cancel(parseError.getStatus().withCause(new CloseWithHeadersMarker()));
+    serverCallTracer.reportCallEnded(false); // error so always false
+  }
+
   /**
    * All of these callbacks are assumed to called on an application thread, and the caller is
    * responsible for handling thrown exceptions.
@@ -337,7 +343,7 @@ private void messagesAvailableInternal(final MessageProducer producer) {
             GrpcUtil.closeQuietly(message);
             GrpcUtil.closeQuietly(producer);
             call.cancelled = true;
-            call.handleInternalError(e);
+            call.handleParseError(e);
             return;
           }
           message.close();