Partially revert "core: change serverimpl,servercallimpl's internalcl…

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

@@ -655,7 +655,8 @@ void setListener(ServerStreamListener listener) {
      * Like {@link ServerCall#close(Status, Metadata)}, but thread-safe for internal use.
      */
     private void internalClose() {
-      stream.cancel(Status.INTERNAL);
+      // TODO(ejona86): this is not thread-safe :)
+      stream.close(Status.UNKNOWN, new Metadata());
     }
 
     @Override

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

@@ -1110,7 +1110,7 @@ public void messageRead_errorCancelsCall() throws Exception {
       fail("Expected exception");
     } catch (TestError t) {
       assertSame(expectedT, t);
-      ensureServerStateIsCancelled();
+      ensureServerStateNotLeaked();
     }
   }
 
@@ -1135,7 +1135,7 @@ public void messageRead_runtimeExceptionCancelsCall() throws Exception {
       fail("Expected exception");
     } catch (RuntimeException t) {
       assertSame(expectedT, t);
-      ensureServerStateIsCancelled();
+      ensureServerStateNotLeaked();
     }
   }
 
@@ -1158,7 +1158,7 @@ public void halfClosed_errorCancelsCall() {
       fail("Expected exception");
     } catch (TestError t) {
       assertSame(expectedT, t);
-      ensureServerStateIsCancelled();
+      ensureServerStateNotLeaked();
     }
   }
 
@@ -1181,7 +1181,7 @@ public void halfClosed_runtimeExceptionCancelsCall() {
       fail("Expected exception");
     } catch (RuntimeException t) {
       assertSame(expectedT, t);
-      ensureServerStateIsCancelled();
+      ensureServerStateNotLeaked();
     }
   }
 
@@ -1204,7 +1204,7 @@ public void onReady_errorCancelsCall() {
       fail("Expected exception");
     } catch (TestError t) {
       assertSame(expectedT, t);
-      ensureServerStateIsCancelled();
+      ensureServerStateNotLeaked();
     }
   }
 
@@ -1227,7 +1227,7 @@ public void onReady_runtimeExceptionCancelsCall() {
       fail("Expected exception");
     } catch (RuntimeException t) {
       assertSame(expectedT, t);
-      ensureServerStateIsCancelled();
+      ensureServerStateNotLeaked();
     }
   }
 
@@ -1430,10 +1430,11 @@ private void verifyExecutorsReturned() {
     verifyNoMoreInteractions(executorPool);
   }
 
-  private void ensureServerStateIsCancelled() {
-    verify(stream).cancel(statusCaptor.capture());
-    assertEquals(Status.INTERNAL, statusCaptor.getValue());
+  private void ensureServerStateNotLeaked() {
+    verify(stream).close(statusCaptor.capture(), metadataCaptor.capture());
+    assertEquals(Status.UNKNOWN, statusCaptor.getValue());
     assertNull(statusCaptor.getValue().getCause());
+    assertTrue(metadataCaptor.getValue().keys().isEmpty());
   }
 
   private static class SimpleServer implements io.grpc.internal.InternalServer {

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

@@ -247,7 +247,7 @@ public void onCompleted() {
         .onNext(StreamingInputCallRequest.getDefaultInstance());
 
     assertTrue(finishLatch.await(900, TimeUnit.MILLISECONDS));
-    assertEquals(Status.CANCELLED.getCode(), Status.fromThrowable(throwableRef.get()).getCode());
+    assertEquals(Status.UNKNOWN, Status.fromThrowable(throwableRef.get()));
     assertNull(responseRef.get());
   }
 }

netty/src/main/java/io/grpc/netty/NettyServerHandler.java

@@ -651,13 +651,8 @@ private void cancelStream(ChannelHandlerContext ctx, CancelServerStreamCommand c
       ChannelPromise promise) {
     // Notify the listener if we haven't already.
     cmd.stream().transportReportStatus(cmd.reason());
-    Http2Error http2Error = Http2Error.INTERNAL_ERROR;
-    if (Status.DEADLINE_EXCEEDED.getCode().equals(cmd.reason().getCode()) || Status.CANCELLED
-        .getCode().equals(cmd.reason().getCode())) {
-      http2Error = Http2Error.CANCEL;
-    }
     // Terminate the stream.
-    encoder().writeRstStream(ctx, cmd.stream().id(), http2Error.code(), promise);
+    encoder().writeRstStream(ctx, cmd.stream().id(), Http2Error.CANCEL.code(), promise);
   }
 
   private void forcefulClose(final ChannelHandlerContext ctx, final ForcefulCloseCommand msg,