Avoid flushing the stream after writing a message in case the server returns a single response

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

@@ -169,7 +169,9 @@ private void sendMessageInternal(RespT message) {
     try {
       InputStream resp = method.streamResponse(message);
       stream.writeMessage(resp);
-      stream.flush();
+      if (!getMethodDescriptor().getType().serverSendsOneMessage()) {
+        stream.flush();
+      }
     } catch (RuntimeException e) {
       close(Status.fromThrowable(e), new Metadata());
     } catch (Error e) {

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

@@ -189,7 +189,6 @@ public void sendMessage() {
     call.sendMessage(1234L);
 
     verify(stream).writeMessage(isA(InputStream.class));
-    verify(stream).flush();
   }
 
   @Test

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

@@ -347,10 +347,10 @@ public void statsRecorded() throws Exception {
     fakeClock.forwardTime(2, SECONDS);
     serverCall.sendHeaders(new Metadata());
     serverCall.sendMessage(3);
+    serverCall.close(Status.OK, new Metadata());
     call.request(1);
     assertInboundMessageRecorded();
     assertInboundWireSizeRecorded(1);
-    serverCall.close(Status.OK, new Metadata());
     assertRpcStatusRecorded(Status.Code.OK, 2000, 2);
     assertRetryStatsRecorded(1, 0, 10_000);
   }