okhttp: Skip trash data for finished stream.

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

@@ -38,11 +38,15 @@ import io.grpc.ManagedChannel;
 import io.grpc.netty.GrpcSslContexts;
 import io.grpc.netty.NettyServerBuilder;
 import io.grpc.okhttp.OkHttpChannelBuilder;
+import io.grpc.stub.StreamObserver;
+import io.grpc.testing.StreamRecorder;
 import io.grpc.testing.TestUtils;
+
 import io.netty.handler.ssl.SupportedCipherSuiteFilter;
 
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
+import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.JUnit4;
 
@@ -90,4 +94,27 @@ public class Http2OkHttpTest extends AbstractTransportTest {
     }
     return builder.build();
   }
+
+  @Test(timeout = 10000)
+  public void receivedDataForFinishedStream() throws Exception {
+    Messages.ResponseParameters.Builder responseParameters =
+        Messages.ResponseParameters.newBuilder()
+        .setSize(1);
+    Messages.StreamingOutputCallRequest.Builder requestBuilder =
+        Messages.StreamingOutputCallRequest.newBuilder()
+            .setResponseType(Messages.PayloadType.COMPRESSABLE);
+    for (int i = 0; i < 10000; i++) {
+      requestBuilder.addResponseParameters(responseParameters);
+    }
+
+    StreamRecorder<Messages.StreamingOutputCallResponse> recorder = StreamRecorder.create();
+    StreamObserver<Messages.StreamingOutputCallRequest> requestStream =
+        asyncStub.fullDuplexCall(recorder);
+    requestStream.onNext(requestBuilder.build());
+    recorder.firstValue().get();
+    requestStream.onError(new Exception("failed"));
+
+    recorder.awaitCompletion();
+    emptyUnary();
+  }
 }

okhttp/src/main/java/io/grpc/okhttp/OkHttpClientTransport.java

@@ -614,6 +614,7 @@ class OkHttpClientTransport implements ClientTransport {
       if (stream == null) {
         if (mayHaveCreatedStream(streamId)) {
           frameWriter.rstStream(streamId, ErrorCode.INVALID_STREAM);
+          in.skip(length);
         } else {
           onError(ErrorCode.PROTOCOL_ERROR, "Received data for unknown stream: " + streamId);
           return;