trying to improve stability of NettyFlowControlTest

Author: panchenko

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

@@ -0,0 +1,38 @@
+/*
+ * Copyright 2025 The gRPC Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.grpc.internal;
+
+import java.util.concurrent.CountDownLatch;
+
+/**
+ * An extension of {@link java.util.concurrent.CountDownLatch CountDownLatch} to simplify usage when
+ * there are only 2 states.
+ */
+public class BooleanLatch extends CountDownLatch {
+
+  public BooleanLatch() {
+    super(1);
+  }
+
+  public boolean isSignaled() {
+    return getCount() == 0;
+  }
+
+  public void signal() {
+    countDown();
+  }
+}

interop-testing/src/test/java/io/grpc/testing/integration/NettyFlowControlTest.java

@@ -24,6 +24,7 @@
 import io.grpc.ServerBuilder;
 import io.grpc.ServerInterceptor;
 import io.grpc.ServerInterceptors;
+import io.grpc.internal.BooleanLatch;
 import io.grpc.netty.GrpcHttp2ConnectionHandler;
 import io.grpc.netty.InternalNettyChannelBuilder;
 import io.grpc.netty.InternalNettyChannelBuilder.ProtocolNegotiatorFactory;
@@ -41,7 +42,6 @@
 import io.netty.util.AsciiString;
 import java.io.IOException;
 import java.net.InetSocketAddress;
-import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
@@ -190,10 +190,9 @@ private void startServer(int serverFlowControlWindow) {
   private static class TestStreamObserver implements StreamObserver<StreamingOutputCallResponse> {
 
     final AtomicReference<GrpcHttp2ConnectionHandler> grpcHandlerRef;
-    final CountDownLatch latch = new CountDownLatch(1);
+    final BooleanLatch completed = new BooleanLatch();
     final long expectedWindow;
-    int lastWindow;
-    boolean wasCompleted;
+    volatile int lastWindow;
 
     public TestStreamObserver(
         AtomicReference<GrpcHttp2ConnectionHandler> grpcHandlerRef, long window) {
@@ -208,31 +207,30 @@ public void onNext(StreamingOutputCallResponse value) {
       int curWindow = grpcHandler.decoder().flowController().initialWindowSize(connectionStream);
       synchronized (this) {
         if (curWindow >= expectedWindow) {
-          if (wasCompleted) {
+          if (completed.isSignaled()) {
             return;
           }
-          wasCompleted = true;
           lastWindow = curWindow;
-          onCompleted();
-        } else if (!wasCompleted) {
+          completed.signal();
+        } else if (!completed.isSignaled()) {
           lastWindow = curWindow;
         }
       }
     }
 
     @Override
     public void onError(Throwable t) {
-      latch.countDown();
+      completed.signal();
       throw new RuntimeException(t);
     }
 
     @Override
     public void onCompleted() {
-      latch.countDown();
+      completed.signal();
     }
 
     public int waitFor(long duration, TimeUnit unit) throws InterruptedException {
-      latch.await(duration, unit);
+      assertTrue("should be completed", completed.await(duration, unit));
       return lastWindow;
     }
   }