Add unit tests for JTATSSL triggerEvent and cover closed stream event behavior.

- Added unit tests in ServerImplTest for JumpToApplicationThreadServerStreamListener.triggerEvent.
- Added serverStream_triggerEvent_afterClose in AbstractTransportTest to verify events are ignored after stream closure.
- Updated Inbound.ServerInbound to check isClosed() before triggering events.

TAG=agy
CONV=e1bfa5a2-e855-4f79-abdd-ef2b264977be

Author: kannanjgithub

binder/src/main/java/io/grpc/binder/internal/Inbound.java

@@ -671,6 +671,9 @@ protected void deliverCloseAbnormal(Status status) {
     void triggerEvent(Object event) {
       ServerStreamListener localListener;
       synchronized (this) {
+        if (isClosed()) {
+          return;
+        }
         localListener = listener;
       }
       if (localListener != null) {

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

@@ -1689,6 +1689,52 @@ public void onReady_runtimeExceptionCancelsCall() {
     }
   }
 
+  @Test
+  public void triggerEvent_delegatesToListener() {
+    JumpToApplicationThreadServerStreamListener listener
+        = new JumpToApplicationThreadServerStreamListener(
+            executor.getScheduledExecutorService(),
+            executor.getScheduledExecutorService(),
+            stream,
+            Context.ROOT.withCancellation(),
+            PerfMark.createTag());
+    ServerStreamListener mockListener = mock(ServerStreamListener.class);
+    listener.setListener(mockListener);
+
+    Object event = new Object();
+    listener.triggerEvent(event);
+
+    verify(mockListener, never()).triggerEvent(any());
+
+    executor.runDueTasks();
+    verify(mockListener).triggerEvent(event);
+  }
+
+  @Test
+  public void triggerEvent_errorCancelsCall() {
+    JumpToApplicationThreadServerStreamListener listener
+        = new JumpToApplicationThreadServerStreamListener(
+            executor.getScheduledExecutorService(),
+            executor.getScheduledExecutorService(),
+            stream,
+            Context.ROOT.withCancellation(),
+            PerfMark.createTag());
+    ServerStreamListener mockListener = mock(ServerStreamListener.class);
+    listener.setListener(mockListener);
+
+    TestError expectedT = new TestError();
+    doThrow(expectedT).when(mockListener).triggerEvent(any());
+
+    listener.triggerEvent(new Object());
+    try {
+      executor.runDueTasks();
+      fail("Expected exception");
+    } catch (TestError t) {
+      assertSame(expectedT, t);
+      ensureServerStateNotLeaked();
+    }
+  }
+
   @Test
   public void binaryLogInstalled() throws Exception {
     final SettableFuture<Boolean> intercepted = SettableFuture.create();

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

@@ -2117,6 +2117,36 @@ public void serverStream_triggerEvent() throws Exception {
     clientStream.cancel(Status.CANCELLED);
   }
 
+  @Test
+  public void serverStream_triggerEvent_afterClose() throws Exception {
+    server.start(serverListener);
+    client = newClientTransport(server);
+    startTransport(client, mockClientTransportListener);
+    MockServerTransportListener serverTransportListener
+        = serverListener.takeListenerOrFail(TIMEOUT_MS, TimeUnit.MILLISECONDS);
+    serverTransport = serverTransportListener.transport;
+
+    ClientStream clientStream = client.newStream(
+        methodDescriptor, new Metadata(), callOptions, noopTracers);
+    ClientStreamListenerBase clientStreamListener = new ClientStreamListenerBase();
+    clientStream.start(clientStreamListener);
+
+    StreamCreation serverStreamCreation
+        = serverTransportListener.takeStreamOrFail(TIMEOUT_MS, TimeUnit.MILLISECONDS);
+    ServerStream serverStream = serverStreamCreation.stream;
+    ServerStreamListenerBase serverStreamListener = serverStreamCreation.listener;
+
+    // Close the stream from client side
+    clientStream.cancel(Status.CANCELLED);
+
+    Object event = new Object();
+    serverStream.triggerEvent(event);
+
+    // Verify listener did NOT receive the event
+    Object receivedEvent = serverStreamListener.eventQueue.poll(100, TimeUnit.MILLISECONDS);
+    assertNull(receivedEvent);
+  }
+
   /**
    * Helper that simply does an RPC. It can be used similar to a sleep for negative testing: to give
    * time for actions _not_ to happen. Since it is based on doing an actual RPC with actual