Fix flaky [Http|Grpc]LifecycleObserverTest (#2376)

Motivation:

It's possible that requesting more data from the payload publisher can
race with completion. Subsequent `request(N)` are possible even after
both observers terminate.

Modifications:

- Await for client & server shutdown before starting verification of
invoked observer callbacks.

Result:

No more events are possible after server shutdown. Verification step
sees all observed events.

Resolves #2343.
Resolves #2298.

servicetalk-grpc-netty/src/test/java/io/servicetalk/grpc/netty/GrpcLifecycleObserverTest.java

@@ -173,10 +173,12 @@ void tearDown() throws Exception {
         try {
             if (client != null) {
                 client.close();
+                client = null;
             }
         } finally {
             if (server != null) {
                 server.close();
+                server = null;
             }
         }
     }
@@ -215,7 +217,10 @@ private void runTest(Callable<String> executeRequest, boolean error, boolean agg
             assertThat(executeRequest.call(), equalTo(CONTENT));
         }
 
+        // Await full termination to make sure no more callbacks will be invoked.
         bothTerminate.await();
+        tearDown();
+
         verifyObservers(true, error, aggregated, clientLifecycleObserver, clientExchangeObserver,
                 clientRequestObserver, clientResponseObserver, clientInOrder, clientRequestInOrder);
         verifyObservers(false, error, aggregated, serverLifecycleObserver, serverExchangeObserver,

servicetalk-http-netty/src/test/java/io/servicetalk/http/netty/HttpLifecycleObserverTest.java

@@ -149,7 +149,7 @@ void testCompleteEmptyMessageBody(HttpProtocol protocol) throws Exception {
         setUp(protocol);
         makeRequestAndAssertResponse(SVC_NO_CONTENT, protocol, NO_CONTENT, 0);
 
-        bothTerminate.await();
+        awaitFullTermination();
         verifyObservers(true, clientLifecycleObserver, clientExchangeObserver, clientRequestObserver,
                 clientResponseObserver, clientInOrder, clientRequestInOrder, false, false, 0);
         verifyObservers(false, serverLifecycleObserver, serverExchangeObserver, serverRequestObserver,
@@ -162,7 +162,7 @@ void testCompleteWithPayloadBodyAndTrailers(HttpProtocol protocol) throws Except
         setUp(protocol);
         makeRequestAndAssertResponse(SVC_ECHO, protocol, OK, CONTENT.readableBytes());
 
-        bothTerminate.await();
+        awaitFullTermination();
         verifyObservers(true, clientLifecycleObserver, clientExchangeObserver, clientRequestObserver,
                 clientResponseObserver, clientInOrder, clientRequestInOrder, true, true,
                 CONTENT.readableBytes());
@@ -177,7 +177,7 @@ void testServerThrows(HttpProtocol protocol) throws Exception {
         setUp(protocol);
         makeRequestAndAssertResponse(SVC_THROW_ERROR, protocol, INTERNAL_SERVER_ERROR, 0);
 
-        bothTerminate.await();
+        awaitFullTermination();
         verifyObservers(true, clientLifecycleObserver, clientExchangeObserver, clientRequestObserver,
                 clientResponseObserver, clientInOrder, clientRequestInOrder, false, false, 0);
         verifyObservers(false, serverLifecycleObserver, serverExchangeObserver, serverRequestObserver,
@@ -190,7 +190,7 @@ void testServerSingleFailed(HttpProtocol protocol) throws Exception {
         setUp(protocol);
         makeRequestAndAssertResponse(SVC_SINGLE_ERROR, protocol, INTERNAL_SERVER_ERROR, 0);
 
-        bothTerminate.await();
+        awaitFullTermination();
         verifyObservers(true, clientLifecycleObserver, clientExchangeObserver, clientRequestObserver,
                 clientResponseObserver, clientInOrder, clientRequestInOrder, false, false, 0);
         verifyObservers(false, serverLifecycleObserver, serverExchangeObserver, serverRequestObserver,
@@ -206,7 +206,7 @@ void testServerPayloadBodyFailure(HttpProtocol protocol) throws Exception {
         assertThat(e.getCause(), instanceOf(protocol == HttpProtocol.HTTP_2 ?
                 Http2Exception.class : ClosedChannelException.class));
 
-        bothTerminate.await();
+        awaitFullTermination();
         verifyError(true, clientLifecycleObserver, clientExchangeObserver, clientRequestObserver,
                 clientResponseObserver, clientInOrder, clientRequestInOrder, 0);
         verifyError(false, serverLifecycleObserver, serverExchangeObserver, serverRequestObserver,
@@ -229,7 +229,7 @@ public Single<StreamingHttpResponse> request(StreamingHttpRequest request) {
         assertThat(e.getCause(), sameInstance(DELIBERATE_EXCEPTION));
 
         bothTerminate.countDown();  // server is not involved in this test, count down manually
-        bothTerminate.await();
+        awaitFullTermination();
 
         clientInOrder.verify(clientLifecycleObserver).onNewExchange();
         clientInOrder.verify(clientExchangeObserver).onRequest(any(StreamingHttpRequest.class));
@@ -264,7 +264,7 @@ public Single<StreamingHttpResponse> handle(HttpServiceContext ctx,
         requestReceived.await();
         responseFuture.cancel(true);
 
-        bothTerminate.await();
+        awaitFullTermination();
 
         verify(serverExchangeObserver, await()).onExchangeFinally();
 
@@ -327,7 +327,7 @@ public Single<StreamingHttpResponse> handle(HttpServiceContext ctx,
         serverResponsePayload.onNext(CONTENT.duplicate());
         serverResponsePayload.onNext(CONTENT.duplicate());
 
-        bothTerminate.await();
+        awaitFullTermination();
 
         clientInOrder.verify(clientLifecycleObserver).onNewExchange();
         clientInOrder.verify(clientExchangeObserver).onConnectionSelected(any(ConnectionInfo.class));
@@ -419,4 +419,10 @@ private static void verifyError(boolean client, HttpLifecycleObserver lifecycle,
         inOrder.verify(exchange).onExchangeFinally();
         verifyNoMoreInteractions(request, response, exchange, lifecycle);
     }
+
+    private void awaitFullTermination() throws Exception {
+        bothTerminate.await();
+        stopServer();
+        // No more callbacks will be invoked after this point.
+    }
 }