Repository: grpc/grpc-java
Issue: Double-closure of call during interop tests

I find it in the stderr of the test report of Netty, Netty local channel, and OkHttp. I see it printed out to my console once, so I think that may be from the InProcess test.

This exception very likely means we have a bug and are double-closing.

```
java.lang.IllegalStateException: call already closed
    at com.google.common.base.Preconditions.checkState(Preconditions.java:173)
    at io.grpc.ServerImpl$ServerCallImpl.close(ServerImpl.java:507)
    at io.grpc.ForwardingServerCall.close(ForwardingServerCall.java:65)
    at io.grpc.testing.TestUtils$1$1.close(TestUtils.java:109)
    at io.grpc.stub.ServerCalls$ResponseObserver.onError(ServerCalls.java:237)
    at io.grpc.testing.integration.TestServiceImpl$ResponseDispatcher.dispatchChunk(TestServiceImpl.java:277)
    at io.grpc.testing.integration.TestServiceImpl$ResponseDispatcher.access$000(TestServiceImpl.java:207)
    at io.grpc.testing.integration.TestServiceImpl$ResponseDispatcher$1.run(TestServiceImpl.java:219)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
    at java.util.concurrent.FutureTask.run(FutureTask.java:266)
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)
```


Changed files:
interop-testing/src/main/java/io/grpc/testing/integration/TestServiceImpl.java, stub/src/main/java/io/grpc/stub/ServerCalls.java

diff --git a/interop-testing/src/main/java/io/grpc/testing/integration/TestServiceImpl.java b/interop-testing/src/main/java/io/grpc/testing/integration/TestServiceImpl.java
index b3de8b624..d73aee095 100644
--- a/interop-testing/src/main/java/io/grpc/testing/integration/TestServiceImpl.java
+++ b/interop-testing/src/main/java/io/grpc/testing/integration/TestServiceImpl.java
@@ -35,6 +35,7 @@ import com.google.common.collect.Queues;
 import com.google.protobuf.ByteString;
 import com.google.protobuf.EmptyProtos;
 
+import io.grpc.Status;
 import io.grpc.stub.StreamObserver;
 import io.grpc.testing.integration.Messages.PayloadType;
 import io.grpc.testing.integration.Messages.ResponseParameters;
@@ -205,10 +206,11 @@ public class TestServiceImpl implements TestServiceGrpc.TestService {
    * available, the stream is half-closed.
    */
   private class ResponseDispatcher {
+    private final Chunk completionChunk = new Chunk(0, 0, 0, false);
     private final Queue<Chunk> chunks;
     private final StreamObserver<StreamingOutputCallResponse> responseStream;
-    private volatile boolean isInputComplete;
     private boolean scheduled;
+    private Throwable failure;
     private Runnable dispatchTask = new Runnable() {
       @Override
       public void run() {
@@ -247,6 +249,7 @@ public class TestServiceImpl implements TestServiceGrpc.TestService {
      * needed.
      */
     public synchronized ResponseDispatcher enqueue(Queue<Chunk> moreChunks) {
+      assertNotFailed();
       chunks.addAll(moreChunks);
       scheduleNextChunk();
       return this;
@@ -257,7 +260,8 @@ public class TestServiceImpl implements TestServiceGrpc.TestService {
      * remain to be scheduled for dispatch to the client.
      */
     public ResponseDispatcher completeInput() {
-      isInputComplete = true;
+      assertNotFailed();
+      chunks.add(completionChunk);
       scheduleNextChunk();
       return this;
     }
@@ -266,15 +270,24 @@ public class TestServiceImpl implements TestServiceGrpc.TestService {
      * Dispatches the current response chunk to the client. This is only called by the executor. At
      * any time, a given dispatch task should only be registered with the executor once.
      */
-    private void dispatchChunk() {
+    private synchronized void dispatchChunk() {
       try {
-
         // Pop off the next chunk and send it to the client.
         Chunk chunk = chunks.remove();
-        responseStream.onNext(chunk.toResponse());
-
+        if (chunk == completionChunk) {
+          responseStream.onCompleted();
+        } else {
+          responseStream.onNext(chunk.toResponse());
+        }
       } catch (Throwable e) {
-        responseStream.onError(e);
+        failure = e;
+        if (Status.fromThrowable(e).getCode() == Status.CANCELLED.getCode()) {
+          // Stream was cancelled by client, responseStream.onError() might be called already or
+          // will be called soon by inbounding StreamObserver.
+          chunks.clear();
+        } else {
+          responseStream.onError(e);
+        }
       }
     }
 
@@ -297,14 +310,12 @@ public class TestServiceImpl implements TestServiceGrpc.TestService {
           return;
         }
       }
+    }
 
-      if (isInputComplete) {
-        // All of the response chunks have been enqueued but there is no chunks left. Close the
-        // stream.
-        responseStream.onCompleted();
+    private void assertNotFailed() {
+      if (failure != null) {
+        throw new IllegalStateException("Stream already failed", failure);
       }
-
-      // Otherwise, the input is not yet complete - wait for more chunks to arrive.
     }
   }
 
diff --git a/stub/src/main/java/io/grpc/stub/ServerCalls.java b/stub/src/main/java/io/grpc/stub/ServerCalls.java
index 9cf44508a..dc2a46044 100644
--- a/stub/src/main/java/io/grpc/stub/ServerCalls.java
+++ b/stub/src/main/java/io/grpc/stub/ServerCalls.java
@@ -202,10 +202,10 @@ public class ServerCalls {
 
           @Override
           public void onCancel() {
+            responseObserver.cancelled = true;
             if (!halfClosed) {
               requestObserver.onError(Status.CANCELLED.asException());
             }
-            responseObserver.cancelled = true;
           }
         };
       }