core: delay retriable stream master listener close until all sub streams are closed

[YifeiZhuang]

This is follow up fix of #9626

This helps to prevent retryable stream from using calloptions.executor when it shouldn't. It is done by delaying notifying upper layer (through masterListener).

[YifeiZhuang]

A very good test case (w/ debug message):

    Dec 15, 2022 5:58:51 PM io.grpc.internal.RetriableStream createSubstream
    INFO: created sub stream, inflight count  1
    Dec 15, 2022 5:58:52 PM io.grpc.internal.RetriableStream$1CommitTask run
    INFO: cancelling drained : {delegate=io.grpc.netty.NettyClientStream@414be180}
    Dec 15, 2022 5:58:52 PM io.grpc.internal.RetriableStream safeCloseMasterListener
    INFO: save close reason status:Status{code=DEADLINE_EXCEEDED, description=deadline exceeded after 0.959140709s. [closed=[], open=[[remote_addr=/0:0:0:0:0:0:0:0:49769]]], cause=null}, progress PROCESSED
io.grpc.testing.integration.AutoWindowSizingOnTest > deadlineExceeded FAILED
    org.junit.runners.model.TestTimedOutException: test timed out after 15 seconds
        at sun.misc.Unsafe.park(Native Method)
        at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)
        at io.grpc.stub.ClientCalls$ThreadlessExecutor.waitAndDrain(ClientCalls.java:748)
        at io.grpc.stub.ClientCalls$BlockingResponseStream.waitForNext(ClientCalls.java:633)
        at io.grpc.stub.ClientCalls$BlockingResponseStream.hasNext(ClientCalls.java:657)
        at io.grpc.stub.ClientCalls$BlockingResponseStream.next(ClientCalls.java:676)
        at io.grpc.testing.integration.AbstractInteropTest.deadlineExceeded(AbstractInteropTest.java:1167)
        at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
        at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
        at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
        at java.lang.reflect.Method.invoke(Method.java:498)
        at org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:50)
        at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
        at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:47)
        at org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
        at org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26)
        at org.junit.internal.runners.statements.RunAfters.evaluate(RunAfters.java:27)
        at org.junit.internal.runners.statements.FailOnTimeout$CallableStatement.call(FailOnTimeout.java:298)
        at org.junit.internal.runners.statements.FailOnTimeout$CallableStatement.call(FailOnTimeout.java:292)
        at java.util.concurrent.FutureTask.run(FutureTask.java:266)
        at java.lang.Thread.run(Thread.java:748)

timeline:
Original stream is created and drained. inFlightStreamCount = 1
RetriableStream is cancelled and noop stream is committed. The commit cancel the original stream. (==> suppose that netty stream will call back on sub stream listener onClose but not happen.)
safeCloseMasterListener is called and listener not closed because inFlightStreamCount is not Integer.MIN_VALUE.

[YifeiZhuang]

RetriableStream is cancelled and noop stream is committed. The commit cancel the original stream. (==> suppose that netty stream will call back on sub stream listener onClose but not happen.)

It turns out the sub listener closed() was called. It looks the race condition that although safeCloseMasterListener happens earlier then subListener.close(), the status not yet saved. So master listener can not be closed, (assert null failed prevents master listener to be called).

[ejona86]

That race fix looks right to me.