test(gax): use FakeApiClock and RecordingScheduler in ScheduledRetryingExecutorTest

sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/core/FakeApiClock.java

@@ -33,9 +33,21 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
 
+/**
+ * A thread-safe fake {@link ApiClock} implementation for unit tests whose time only advances when
+ * {@link #incrementNanoTime(long)} is explicitly called.
+ *
+ * <p>Use this clock in retry and polling tests to decouple timeout and backoff calculations from
+ * real wall-clock time on the test runner.
+ */
 public class FakeApiClock implements ApiClock {
   private final AtomicLong currentNanoTime;
 
+  /**
+   * Creates a fake clock initialized to the given nanosecond timestamp.
+   *
+   * @param initialNanoTime the starting value for {@link #nanoTime()}
+   */
   public FakeApiClock(long initialNanoTime) {
     currentNanoTime = new AtomicLong(initialNanoTime);
   }
@@ -50,6 +62,11 @@ public long millisTime() {
     return TimeUnit.MILLISECONDS.convert(nanoTime(), TimeUnit.NANOSECONDS);
   }
 
+  /**
+   * Advances the clock's current time by {@code delta} nanoseconds.
+   *
+   * @param delta the number of nanoseconds to add to the current time
+   */
   public void incrementNanoTime(long delta) {
     currentNanoTime.getAndAdd(delta);
   }

sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/core/RecordingScheduler.java

@@ -29,79 +29,74 @@
  */
 package com.google.api.gax.core;
 
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.ArgumentMatchers.anyLong;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
-
-import java.util.ArrayList;
-import java.util.List;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
-import org.mockito.Mockito;
-import org.mockito.invocation.InvocationOnMock;
-import org.mockito.stubbing.Answer;
-
-public abstract class RecordingScheduler implements ScheduledExecutorService {
-
-  public abstract List<java.time.Duration> getSleepDurations();
-
-  public abstract int getIterationsCount();
-
-  public static RecordingScheduler create(final FakeApiClock clock) {
-    RecordingScheduler mock =
-        mock(RecordingScheduler.class, Mockito.withSettings().withoutAnnotations());
 
-    // mock class fields:
-    final ScheduledExecutorService executor = new ScheduledThreadPoolExecutor(1);
-    final List<java.time.Duration> sleepDurations = new ArrayList<>();
-    final AtomicInteger iterationsCount = new AtomicInteger(0);
+/**
+ * A test {@link ScheduledExecutorService} that records scheduled task iterations and advances a
+ * {@link FakeApiClock} instead of sleeping in real time.
+ *
+ * <p>When {@link #schedule(Runnable, long, TimeUnit)} is called, this scheduler increments the
+ * iteration count, advances the {@link FakeApiClock} by the requested delay, and immediately
+ * submits the task with zero delay to the underlying single-threaded executor. This allows
+ * multi-attempt retry and polling tests to execute without real-time backoff sleeps while still
+ * exercising simulated elapsed time calculations and preserving asynchronous callback ordering on
+ * {@code CallbackChainRetryingFuture}.
+ */
+public class RecordingScheduler extends ScheduledThreadPoolExecutor {
+  /**
+   * A single worker thread executes zero-delay scheduled tasks sequentially in FIFO order. Running
+   * tasks asynchronously on a dedicated worker thread avoids lock-inversion and reentrant race
+   * conditions when registering attempt futures in CallbackChainRetryingFuture.
+   */
+  private static final int CORE_POOL_SIZE = 1;
 
-    // mock class methods:
-    // ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit);
-    when(mock.schedule(any(Runnable.class), anyLong(), any(TimeUnit.class)))
-        .then(
-            new Answer<ScheduledFuture<?>>() {
-              @Override
-              public ScheduledFuture<?> answer(InvocationOnMock invocation) throws Throwable {
-                Object[] args = invocation.getArguments();
-                Runnable runnable = (Runnable) args[0];
-                Long delay = (Long) args[1];
-                TimeUnit unit = (TimeUnit) args[2];
-                iterationsCount.incrementAndGet();
-                sleepDurations.add(
-                    java.time.Duration.ofMillis(TimeUnit.MILLISECONDS.convert(delay, unit)));
-                clock.incrementNanoTime(TimeUnit.NANOSECONDS.convert(delay, unit));
-                return executor.schedule(runnable, 0, TimeUnit.NANOSECONDS);
-              }
-            });
+  private final FakeApiClock clock;
+  private final AtomicInteger iterationsCount = new AtomicInteger(0);
 
-    // List<Runnable> shutdownNow()
-    when(mock.shutdownNow())
-        .then(
-            new Answer<List<Runnable>>() {
-              @Override
-              public List<Runnable> answer(InvocationOnMock invocation) throws Throwable {
-                return executor.shutdownNow();
-              }
-            });
+  private RecordingScheduler(FakeApiClock clock) {
+    super(CORE_POOL_SIZE);
+    this.clock = clock;
+  }
 
-    // List<java.time.Duration> getSleepDurations()
-    when(mock.getSleepDurations()).thenReturn(sleepDurations);
+  /**
+   * Creates a new {@link RecordingScheduler} bound to the given {@link FakeApiClock}.
+   *
+   * @param clock the fake clock to advance whenever a delayed task is scheduled
+   */
+  public static RecordingScheduler create(FakeApiClock clock) {
+    return new RecordingScheduler(clock);
+  }
 
-    // int getIterationsCount()
-    when(mock.getIterationsCount())
-        .then(
-            new Answer<Integer>() {
-              @Override
-              public Integer answer(InvocationOnMock invocation) throws Throwable {
-                return iterationsCount.get();
-              }
-            });
+  /**
+   * Schedules a task by incrementing the iteration count, advancing the {@link FakeApiClock} by the
+   * requested delay, and immediately submitting the task with zero delay to the worker thread.
+   *
+   * <p>Advancing the fake clock simulates the passage of backoff time deterministically without
+   * sleeping in real time. Submitting with zero delay to the worker thread preserves asynchronous
+   * execution so that attempt registration in {@code CallbackChainRetryingFuture} completes cleanly
+   * before the scheduled attempt runs.
+   *
+   * <p>Only the {@link Runnable} overload of {@code schedule} needs to be overridden because {@code
+   * ScheduledRetryingExecutor} wraps its scheduler with {@code
+   * MoreExecutors.listeningDecorator(scheduler)}, which wraps {@link java.util.concurrent.Callable}
+   * tasks in a {@code TrustedListenableFutureTask} (a {@link Runnable}) before delegating to this
+   * method.
+   */
+  @Override
+  public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) {
+    iterationsCount.incrementAndGet();
+    clock.incrementNanoTime(TimeUnit.NANOSECONDS.convert(delay, unit));
+    return super.schedule(command, 0, TimeUnit.NANOSECONDS);
+  }
 
-    return mock;
+  /**
+   * Returns the total number of tasks scheduled via {@link #schedule(Runnable, long, TimeUnit)}.
+   */
+  public int getIterationsCount() {
+    return iterationsCount.get();
   }
 }

sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/retrying/ScheduledRetryingExecutorTest.java

@@ -38,8 +38,10 @@
 import static org.junit.jupiter.api.Assertions.assertSame;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
+import com.google.api.core.ApiClock;
 import com.google.api.core.ApiFuture;
-import com.google.api.core.NanoClock;
+import com.google.api.gax.core.FakeApiClock;
+import com.google.api.gax.core.RecordingScheduler;
 import com.google.api.gax.retrying.FailingCallable.CustomException;
 import com.google.api.gax.rpc.testing.FakeCallContext;
 import java.time.Duration;
@@ -51,25 +53,48 @@
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.mockito.Mockito;
 
 class ScheduledRetryingExecutorTest extends AbstractRetryingExecutorTest {
 
   // Number of test runs, essential for multithreaded tests.
   private static final int EXECUTIONS_COUNT = 5;
+  private FakeApiClock fakeClock;
+  private RecordingScheduler recordingScheduler;
+
+  @BeforeEach
+  void setUp() {
+    // ExponentialRetryAlgorithm and ExponentialPollAlgorithm only compute relative elapsed time
+    // deltas against firstAttemptStartTimeNanos (clock.nanoTime() - firstAttemptStartTimeNanos)
+    // and never treat 0L as a special sentinel value.
+    fakeClock = new FakeApiClock(0L);
+    recordingScheduler = RecordingScheduler.create(fakeClock);
+  }
+
+  @AfterEach
+  void tearDown() {
+    recordingScheduler.shutdownNow();
+  }
+
+  @Override
+  protected ApiClock getClock() {
+    return fakeClock;
+  }
 
   @Override
   protected RetryingExecutorWithContext<String> getExecutor(RetryAlgorithm<String> retryAlgorithm) {
-    return getRetryingExecutor(retryAlgorithm, scheduledExecutorService);
+    return getRetryingExecutor(retryAlgorithm, recordingScheduler);
   }
 
   @Override
   protected RetryAlgorithm<String> getAlgorithm(
       RetrySettings retrySettings, int apocalypseCountDown, RuntimeException apocalypseException) {
     return new RetryAlgorithm<>(
         new TestResultRetryAlgorithm<String>(apocalypseCountDown, apocalypseException),
-        new ExponentialRetryAlgorithm(retrySettings, NanoClock.getDefaultClock()));
+        new ExponentialRetryAlgorithm(retrySettings, fakeClock));
   }
 
   private RetryingExecutorWithContext<String> getRetryingExecutor(
@@ -145,7 +170,7 @@ void testSuccessWithFailuresGetAttempt() throws Exception {
       FailingCallable callable = new FailingCallable(15, "request", "SUCCESS", tracer);
 
       RetryingExecutorWithContext<String> executor =
-          getRetryingExecutor(getAlgorithm(retrySettings, 0, null), scheduledExecutorService);
+          getRetryingExecutor(getAlgorithm(retrySettings, 0, null), recordingScheduler);
       RetryingFuture<String> future =
           executor.createFuture(
               callable,