Fix racy response leak check in gRPC ProtocolCompatibilityTest (#3360)

Motivation:

#3354 introduced `ProtocolCompatibilityTest.ResponseLeakValidator`.
However, the test thread that asserts `pendingRequests` counter can race
with `GrpcUtils.validateResponseAndGetPayload` logic for Trailers-Only
response draining that executes inside `afterOnError` callback.

Modifications:

- Use `CountDownLatch` to ensure all new requests were terminated.

Result:

Fixes #3356

Author: idelpivnitskiy

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

@@ -117,9 +117,13 @@
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 import java.util.Queue;
 import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -214,7 +218,7 @@ public SocketAddress listenAddress() {
     private final ResponseLeakValidator responseLeakValidator = new ResponseLeakValidator();
 
     @AfterEach
-    void finalChecks() {
+    void finalChecks() throws Exception {
         responseLeakValidator.assertNoPendingRequests();
     }
 
@@ -1876,7 +1880,8 @@ private CompatResponse response(final int value) throws Exception {
 
     private static final class ResponseLeakValidator implements StreamingHttpClientFilterFactory {
 
-        private final AtomicInteger pendingRequests = new AtomicInteger();
+        private final AtomicInteger requestCounter = new AtomicInteger();
+        private final ConcurrentMap<String, CountDownLatch> pendingRequests = new ConcurrentHashMap<>();
 
         @Override
         public StreamingHttpClientFilter create(FilterableStreamingHttpClient client) {
@@ -1885,9 +1890,11 @@ public StreamingHttpClientFilter create(FilterableStreamingHttpClient client) {
                 protected Single<StreamingHttpResponse> request(StreamingHttpRequester delegate,
                                                                 StreamingHttpRequest request) {
                     return Single.defer(() -> {
-                        pendingRequests.incrementAndGet();
+                        CountDownLatch requestLatch = new CountDownLatch(1);
+                        pendingRequests.put(requestCounter.incrementAndGet() + ". " + request.requestTarget(),
+                                requestLatch);
                         return delegate.request(request)
-                                .liftSync(new BeforeFinallyHttpOperator(pendingRequests::decrementAndGet))
+                                .liftSync(new BeforeFinallyHttpOperator(requestLatch::countDown))
                                 .shareContextOnSubscribe();
                     });
                 }
@@ -1899,8 +1906,11 @@ public HttpExecutionStrategy requiredOffloads() {
             return HttpExecutionStrategies.offloadNone();
         }
 
-        void assertNoPendingRequests() {
-            assertThat("Detected pending requests, possible response leak", pendingRequests.get(), is(0));
+        void assertNoPendingRequests() throws Exception {
+            for (Map.Entry<String, CountDownLatch> entry : pendingRequests.entrySet()) {
+                assertThat("Request #" + entry.getKey() + "was not properly drained, possible response leak",
+                        entry.getValue().await(5, SECONDS), is(true));
+            }
         }
     }
 }