diff --git a/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/core/FakeApiClock.java b/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/core/FakeApiClock.java index 087fbf1ec3a5..fed37ba6d8f3 100644 --- a/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/core/FakeApiClock.java +++ b/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. + * + *

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); } diff --git a/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/core/RecordingScheduler.java b/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/core/RecordingScheduler.java index fddab80dcbd5..b2e0bf09c805 100644 --- a/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/core/RecordingScheduler.java +++ b/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/core/RecordingScheduler.java @@ -29,11 +29,7 @@ */ 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.time.Duration; import java.util.ArrayList; import java.util.List; import java.util.concurrent.ScheduledExecutorService; @@ -41,67 +37,68 @@ 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 getSleepDurations(); - public abstract int getIterationsCount(); - - public static RecordingScheduler create(final FakeApiClock clock) { - RecordingScheduler mock = - mock(RecordingScheduler.class, Mockito.withSettings().withoutAnnotations()); +/** + * A test {@link ScheduledExecutorService} that records scheduled delays and advances a {@link + * FakeApiClock} instead of sleeping in real time. + * + *

When {@link #schedule(Runnable, long, TimeUnit)} is called, this scheduler records the + * requested delay, increments the {@link FakeApiClock} by that 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 fields: - final ScheduledExecutorService executor = new ScheduledThreadPoolExecutor(1); - final List sleepDurations = new ArrayList<>(); - final AtomicInteger iterationsCount = new AtomicInteger(0); + private final FakeApiClock clock; + private final List sleepDurations = new ArrayList<>(); + private final AtomicInteger iterationsCount = new AtomicInteger(0); - // mock class methods: - // ScheduledFuture schedule(Runnable command, long delay, TimeUnit unit); - when(mock.schedule(any(Runnable.class), anyLong(), any(TimeUnit.class))) - .then( - new Answer>() { - @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 RecordingScheduler(FakeApiClock clock) { + super(CORE_POOL_SIZE); + this.clock = clock; + } - // List shutdownNow() - when(mock.shutdownNow()) - .then( - new Answer>() { - @Override - public List answer(InvocationOnMock invocation) throws Throwable { - return executor.shutdownNow(); - } - }); + /** + * 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); + } - // List getSleepDurations() - when(mock.getSleepDurations()).thenReturn(sleepDurations); + /** + * Schedules a task by recording the requested delay, advancing the {@link FakeApiClock} by that + * duration, and immediately submitting the task with zero delay to the worker thread. + * + *

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. + */ + @Override + public ScheduledFuture schedule(Runnable command, long delay, TimeUnit unit) { + iterationsCount.incrementAndGet(); + sleepDurations.add(Duration.ofMillis(TimeUnit.MILLISECONDS.convert(delay, unit))); + clock.incrementNanoTime(TimeUnit.NANOSECONDS.convert(delay, unit)); + return super.schedule(command, 0, TimeUnit.NANOSECONDS); + } - // int getIterationsCount() - when(mock.getIterationsCount()) - .then( - new Answer() { - @Override - public Integer answer(InvocationOnMock invocation) throws Throwable { - return iterationsCount.get(); - } - }); + /** Returns the retry or poll delays passed to {@link #schedule(Runnable, long, TimeUnit)}. */ + public List getSleepDurations() { + return sleepDurations; + } - return mock; + /** Returns the total number of tasks scheduled via {@link #schedule(Runnable, long, TimeUnit)}. */ + public int getIterationsCount() { + return iterationsCount.get(); } } diff --git a/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/retrying/ScheduledRetryingExecutorTest.java b/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/retrying/ScheduledRetryingExecutorTest.java index 89b807bd5610..12d81f3675b5 100644 --- a/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/retrying/ScheduledRetryingExecutorTest.java +++ b/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,6 +53,8 @@ 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; @@ -58,10 +62,31 @@ 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 getExecutor(RetryAlgorithm retryAlgorithm) { - return getRetryingExecutor(retryAlgorithm, scheduledExecutorService); + return getRetryingExecutor(retryAlgorithm, recordingScheduler); } @Override @@ -69,7 +94,7 @@ protected RetryAlgorithm getAlgorithm( RetrySettings retrySettings, int apocalypseCountDown, RuntimeException apocalypseException) { return new RetryAlgorithm<>( new TestResultRetryAlgorithm(apocalypseCountDown, apocalypseException), - new ExponentialRetryAlgorithm(retrySettings, NanoClock.getDefaultClock())); + new ExponentialRetryAlgorithm(retrySettings, fakeClock)); } private RetryingExecutorWithContext getRetryingExecutor( @@ -145,7 +170,7 @@ void testSuccessWithFailuresGetAttempt() throws Exception { FailingCallable callable = new FailingCallable(15, "request", "SUCCESS", tracer); RetryingExecutorWithContext executor = - getRetryingExecutor(getAlgorithm(retrySettings, 0, null), scheduledExecutorService); + getRetryingExecutor(getAlgorithm(retrySettings, 0, null), recordingScheduler); RetryingFuture future = executor.createFuture( callable,