test: fix concurrency issue in testThrottlingBlocking (#2001)

At the end of the test, it verifies that the batcher waited more than throttled time by capturing the argument of `callContext.withOption()`. There were two problems:

# "Zero interactions with this mock" problem

```
Error:  Failures: 
Error:    BatcherImplTest.testThrottlingBlocking:899 
apiCallContext.withOption(
    <Capturing argument>,
    <Capturing argument>
);
Wanted 1 time:
-> at com.google.api.gax.batching.BatcherImplTest.testThrottlingBlocking(BatcherImplTest.java:899)
But was 0 times.
```

The callContext.withOption method is called by `BatcherImpl.sendOutstanding()` method via
`BatcherImpl$PushCurrentBatchRunnable` in **another thread**. https://togithub.com/googleapis/sdk-platform-java/blob/4c741077d614093d08665e9ddd83fb0e332b7881/gax-java/gax/src/main/java/com/google/api/gax/batching/BatcherImpl.java#L206

Technically, there's no guarantee that the thread calls the withOption method within a certain timeframe, especially when we run the tests concurrently in a machine with many CPU cores (the nightly test setup). That's why we occasionally saw "Zero interactions with this mock" problem. 

In my experiment (https://togithub.com/suztomo/gax-batcher-impl-test-reproducer/blob/main/README.md), increasing 100 ms timeout to 1000 ms just worked good to prevent false positives. The assertion is not about timing, but about BatcherImpl recording the throttled time correctly.

After fixing this problem, then I saw another problem below.

# "expected to be at least: 50 but was: 48" problem

```
[ERROR] com.google.api.gax.batching.BatcherImplTest.testThrottlingBlocking_gax_test06 -- Time elapsed: 5.138 s <<< FAILURE!
expected to be at least: 50
but was                : 48
	at com.google.api.gax.batching.BatcherImplTest.testThrottlingBlocking_gax_test06(BatcherImplTest.java:926)
```

At the end of the test, it verifies that the batcher waited more than throttled time by capturing the argument of `callContext.withOption()`. In ideal, usual scenarios,

1. In thread A, the batcher blocks at `batcher.add` (because of flowController) the thread and starts stopwatch to measure throttled time.
2. In thread B (created by the BatcherImplTest) sleeps 50 ms (`throttledTime`)
3. In thread B, BatcherImplTest calls `flowController.release()`, which wakes up the thread A.
4. The batcher records it was blocked more than 50 ms.

However, in rare cases, these events happens in this order:

1. Thread B starts sleeping 50 ms.
2. In thread A the batcher blocks at `batcher.add` (because of flowController) and starts stopwatch to measure throttled time.
3. In thread B, after 50 ms sleep, BatcherImplTest calls `flowController.release()`, which wakes up the thread A.
5. The batcher records it was blocked **less than** 50 ms. This makes the test assertion fail with "_expected to be at least: 50 but was: 48_" message.

To prevent this rare case, BatcherImplTest needs to wait before sleeping 50 ms in thread B. In this pull request, I add a do-while loop to check the thread A (the thread of `batcher.add`) is in Thread.State.WAITING state before making the sleeping of 50 ms.


- [x] Make sure to open an issue as a [bug/issue](https://togithub.com/googleapis/gapic-generator-java/issues/new/choose) before writing your code! That way we can discuss the change, evaluate designs, and agree on the general idea
- [x] Ensure the tests and linter pass
- [x] Code coverage does not decrease (if any source code was changed)
- [x] Appropriate docs were updated (if necessary)

Fixes #1193  ☕️

Author: suztomo

gax-java/gax/src/test/java/com/google/api/gax/batching/BatcherImplTest.java

@@ -51,6 +51,7 @@
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Queues;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.Queue;
 import java.util.concurrent.Callable;
@@ -830,7 +831,7 @@ public void testConstructors() throws InterruptedException {
     }
   }
 
-  @Test
+  @Test(timeout = 60000)
   public void testThrottlingBlocking() throws Exception {
     BatchingSettings settings =
         BatchingSettings.newBuilder()
@@ -862,16 +863,28 @@ public void testThrottlingBlocking() throws Exception {
             flowController,
             callContext)) {
       flowController.reserve(1, 1);
+      List<Thread> batcherAddThreadHolder = Collections.synchronizedList(new ArrayList<>());
       Future future =
           executor.submit(
               new Runnable() {
                 @Override
                 public void run() {
+                  batcherAddThreadHolder.add(Thread.currentThread());
                   batcher.add(1);
                 }
               });
-      // Add a little delay ensuring that the next step starts after batcher.add(1)
-      Thread.sleep(10);
+
+      // Wait until batcher.add blocks (Thread.State.WAITING) and the batcher starts the
+      // stopwatch for total_throttled_time. Without this proper waiting, the
+      // Thread.sleep(throttledTime) below may start before the stopwatch starts,
+      // resulting in a shorter total_throttled_time at the verification of throttledTime
+      // at the end of the test.
+      // https://github.com/googleapis/sdk-platform-java/issues/1193
+      do {
+        Thread.sleep(10);
+      } while (batcherAddThreadHolder.isEmpty()
+          || batcherAddThreadHolder.get(0).getState() != Thread.State.WAITING);
+
       executor.submit(
           () -> {
             try {
@@ -895,11 +908,18 @@ public void run() {
       }
 
       // Mockito recommends using verify() as the ONLY way to interact with Argument
-      // captors - otherwise it may incur in unexpected behaviour
-      Mockito.verify(callContext, Mockito.timeout(100)).withOption(key.capture(), value.capture());
+      // captors - otherwise it may incur in unexpected behaviour.
+      // The callContext.withOption method is called by BatcherImpl.sendOutstanding() method via
+      // BatcherImpl$PushCurrentBatchRunnable in another thread. Technically, there's no guarantee
+      // that the thread calls the withOption method within a certain timeframe. 1000 ms just works
+      // fine to prevent false positives.
+      // https://github.com/googleapis/sdk-platform-java/issues/1193
+      Mockito.verify(callContext, Mockito.timeout(1000)).withOption(key.capture(), value.capture());
 
       // Verify that throttled time is recorded in ApiCallContext
       assertThat(key.getValue()).isSameInstanceAs(Batcher.THROTTLED_TIME_KEY);
+      // Because this test waited for throttledTime before flowController.release() method,
+      // the recorded total_throttled_time should be higher than or equal to that.
       assertThat(value.getValue()).isAtLeast(throttledTime);
     } finally {
       executor.shutdownNow();