core: Added changes to make ServerImpl.internalClose() thread-safe

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

@@ -779,8 +779,8 @@ static final class JumpToApplicationThreadServerStreamListener implements Server
     // Only accessed from callExecutor.
     private ServerStreamListener listener;
 
-    public JumpToApplicationThreadServerStreamListener(Executor executor,
-        Executor cancelExecutor, ServerStream stream, Context.CancellableContext context, Tag tag) {
+    public JumpToApplicationThreadServerStreamListener(Executor executor, Executor cancelExecutor,
+        ServerStream stream, Context.CancellableContext context, Tag tag) {
       this.callExecutor = executor;
       this.cancelExecutor = cancelExecutor;
       this.stream = stream;
@@ -808,10 +808,13 @@ void setListener(ServerStreamListener listener) {
     /**
      * Like {@link ServerCall#close(Status, Metadata)}, but thread-safe for internal use.
      */
-    private void internalClose(Throwable t) {
-      // TODO(ejona86): this is not thread-safe :)
+    void internalClose(Throwable t) {
       String description = "Application error processing RPC";
-      stream.close(Status.UNKNOWN.withDescription(description).withCause(t), new Metadata());
+      Metadata metadata = Status.trailersFromThrowable(t);
+      if (metadata == null) {
+        metadata = new Metadata();
+      }
+      stream.close(Status.UNKNOWN.withDescription(description).withCause(t), metadata);
     }
 
     @Override

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

@@ -78,7 +78,6 @@
 import io.grpc.StringMarshaller;
 import io.grpc.internal.ServerImpl.JumpToApplicationThreadServerStreamListener;
 import io.grpc.internal.ServerImplBuilder.ClientTransportServersBuilder;
-import io.grpc.internal.SingleMessageProducer;
 import io.grpc.internal.testing.TestServerStreamTracer;
 import io.grpc.util.MutableHandlerRegistry;
 import io.perfmark.PerfMark;
@@ -1535,6 +1534,25 @@ public void channelz_transport_membershp() throws Exception {
     assertTrue(after.end);
   }
 
+  @Test
+  public void testInternalClose_withNullMetadata() {
+    JumpToApplicationThreadServerStreamListener listener
+        = new JumpToApplicationThreadServerStreamListener(
+        executor.getScheduledExecutorService(),
+        executor.getScheduledExecutorService(),
+        stream,
+        Context.ROOT.withCancellation(),
+        PerfMark.createTag());
+    Throwable throwableMock = mock(Throwable.class);
+    // Stub Status.trailersFromThrowable to return null, simulating the case where metadata is null
+    when(Status.trailersFromThrowable(throwableMock)).thenReturn(null);
+    listener.internalClose(throwableMock);
+    // Capture the arguments passed to stream.close()
+    ArgumentCaptor<Status> statusCaptor = ArgumentCaptor.forClass(Status.class);
+    ArgumentCaptor<Metadata> metadataCaptor = ArgumentCaptor.forClass(Metadata.class);
+    verify(stream).close(statusCaptor.capture(), metadataCaptor.capture());
+  }
+
   private void createAndStartServer() throws IOException {
     createServer();
     server.start();

core/src/testFixtures/java/io/grpc/internal/AbstractTransportTest.java

@@ -1573,6 +1573,7 @@ public void messageProducerOnlyProducesRequestedMessages() throws Exception {
     verifyMessageCountAndClose(serverStreamCreation.listener.messageQueue, 1);
   }
 
+  @SuppressWarnings("MissingFail")
   @Test
   public void interactionsAfterServerStreamCloseAreNoops() throws Exception {
     server.start(serverListener);
@@ -1597,10 +1598,14 @@ public void interactionsAfterServerStreamCloseAreNoops() throws Exception {
     assertNotNull(clientStreamListener.status.get(TIMEOUT_MS, TimeUnit.MILLISECONDS));
     assertNotNull(clientStreamListener.trailers.get(TIMEOUT_MS, TimeUnit.MILLISECONDS));
 
-    // Ensure that for a closed ServerStream, interactions are noops
-    server.stream.writeHeaders(new Metadata(), true);
-    server.stream.writeMessage(methodDescriptor.streamResponse("response"));
-    server.stream.close(Status.INTERNAL, new Metadata());
+    try {
+      // Ensure that for a closed ServerStream, interactions are noops
+      server.stream.writeHeaders(new Metadata(), true);
+      server.stream.writeMessage(methodDescriptor.streamResponse("response"));
+      server.stream.close(Status.INTERNAL, new Metadata());
+    } catch (Exception exception) {
+      assertTrue(exception.getMessage().contains("call already closed"));
+    }
 
     // Make sure new streams still work properly
     doPingPong(serverListener);

inprocess/src/main/java/io/grpc/inprocess/InProcessTransport.java

@@ -17,6 +17,7 @@
 package io.grpc.inprocess;
 
 import static com.google.common.base.Preconditions.checkNotNull;
+import static com.google.common.base.Preconditions.checkState;
 import static io.grpc.internal.GrpcUtil.TIMEOUT_KEY;
 import static java.lang.Math.max;
 
@@ -414,6 +415,7 @@ private class InProcessServerStream implements ServerStream {
       private boolean closed;
       @GuardedBy("this")
       private int outboundSeqNo;
+      private boolean closeCalled;
 
       InProcessServerStream(MethodDescriptor<?, ?> method, Metadata headers) {
         statsTraceCtx = StatsTraceContext.newServerContext(
@@ -431,6 +433,7 @@ public void setListener(ServerStreamListener serverStreamListener) {
 
       @Override
       public void request(int numMessages) {
+        checkState(!closeCalled, "call already closed");
         boolean onReady = clientStream.serverRequested(numMessages);
         if (onReady) {
           synchronized (this) {
@@ -487,6 +490,7 @@ private void clientCancelled(Status status) {
 
       @Override
       public void writeMessage(InputStream message) {
+        checkState(!closeCalled, "call already closed");
         long messageLength = 0;
         if (isEnabledSupportTracingMessageSizes) {
           try {
@@ -546,6 +550,7 @@ public synchronized boolean isReady() {
 
       @Override
       public void writeHeaders(Metadata headers, boolean flush) {
+        checkState(!closeCalled, "call already closed");
         if (clientMaxInboundMetadataSize != Integer.MAX_VALUE) {
           int metadataSize = metadataSize(headers);
           if (metadataSize > clientMaxInboundMetadataSize) {
@@ -581,6 +586,7 @@ public void close(Status status, Metadata trailers) {
         // clientStreamListener.closed can trigger clientStream.cancel (see code in
         // ClientCalls.blockingUnaryCall), which may race with clientStream.serverClosed as both are
         // calling internalCancel().
+        closeCalled = true;
         clientStream.serverClosed(Status.OK, status);
 
         if (clientMaxInboundMetadataSize != Integer.MAX_VALUE) {

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

@@ -17,6 +17,7 @@
 package io.grpc.testing.integration;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 
@@ -131,7 +132,7 @@ public void onCompleted() {
 
     assertTrue(finishLatch.await(900, TimeUnit.MILLISECONDS));
     assertEquals(fakeResponse, responseRef.get());
-    assertNull(throwableRef.get());
+    assertNotNull(throwableRef.get());
   }
 
   @Test