testing: Fix race condition in MockServerTransportListener and improve stream cleanup in AbstractTransportTest

In MockServerTransportListener.streamCreated(), stream.setListener(listener)
was called after streams.add(StreamCreation(...)). This created a race
condition where a test thread calling takeStreamOrFail() could dequeue
the stream and call serverStream.triggerEvent() before stream.setListener()
was called by the transport/container thread. When this occurred (e.g. in
TomcatTransportTest on multi-core runners), ServletServerStream invoked
transportState.triggerEvent() on the test thread, saw a null listener,
threw a NullPointerException (swallowed by SerializingExecutor), and
never enqueued the event into the listener queue, leading to a timeout
and assertion failure:
    expected:<...Object@...> but was:<null>

Setting stream.setListener(listener) before enqueuing to streams guarantees
that any thread consuming the StreamCreation will always observe a fully
initialized listener.

Additionally, in AbstractTransportTest.serverStream_triggerEvent(), replace
clientStream.cancel(Status.CANCELLED) with serverStream.close(Status.OK, ...)
for clean stream closure instead of leaving an uncoordinated client RST_STREAM
in flight during container tearDown.

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

Author: kannanjgithub

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

@@ -2114,7 +2114,7 @@ public void serverStream_triggerEvent() throws Exception {
     assertEquals(event, receivedEvent);
 
     // Cleanup
-    clientStream.cancel(Status.CANCELLED);
+    serverStream.close(Status.OK, new Metadata());
   }
 
   @Test

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

@@ -45,8 +45,8 @@ public MockServerTransportListener(ServerTransport transport) {
   @Override
   public void streamCreated(ServerStream stream, String method, Metadata headers) {
     ServerStreamListenerBase listener = new ServerStreamListenerBase();
-    streams.add(new StreamCreation(stream, method, headers, listener));
     stream.setListener(listener);
+    streams.add(new StreamCreation(stream, method, headers, listener));
   }
 
   @Override