Skip to content

test(gax): use FakeApiClock and RecordingScheduler in ScheduledRetryingExecutorTest #14706

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
lqiu96 wants to merge 2 commits into
base: main
Choose a base branch
from
Draft

test(gax): use FakeApiClock and RecordingScheduler in ScheduledRetryingExecutorTest #14706

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
fbc1aaa
test(gax): use FakeApiClock in DirectRetryingExecutorTest to avoid Th…
lqiu96 Oct 20, 2026
049681f
test(gax): simplify ScheduledRetryingExecutorTest and document test h…
lqiu96 Oct 20, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 17 additions & 0 deletions sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/core/FakeApiClock.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
}
Expand All @@ -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);
}
Expand Down
117 changes: 57 additions & 60 deletions sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/core/RecordingScheduler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,79 +29,76 @@
*/
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;
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());
/**
* A test {@link ScheduledExecutorService} that records scheduled delays and advances a {@link
* FakeApiClock} instead of sleeping in real time.
*
* <p>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<java.time.Duration> sleepDurations = new ArrayList<>();
final AtomicInteger iterationsCount = new AtomicInteger(0);
private final FakeApiClock clock;
private final List<Duration> sleepDurations = new ArrayList<>();

@gemini-code-assist gemini-code-assist Bot Oct 20, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

The sleepDurations list is implemented as a standard ArrayList, which is not thread-safe. Since RecordingScheduler extends ScheduledThreadPoolExecutor and can be called from multiple threads concurrently, concurrent calls to schedule can lead to race conditions or ConcurrentModificationException when writing to or reading from sleepDurations.

Using CopyOnWriteArrayList ensures thread safety for both writing (appending new durations) and reading (via getSleepDurations()). If any index-based operations are performed on this list concurrently, remember to snapshot the list to an array first to prevent IndexOutOfBoundsException.

Suggested change
private final List<Duration> sleepDurations = new ArrayList<>();
private final List<Duration> sleepDurations = new java.util.concurrent.CopyOnWriteArrayList<>();
References
  1. Before suggesting thread-safe collection alternatives (such as CopyOnWriteArrayList) to avoid lock contention, verify the existing definition of the collection to ensure it is not already thread-safe.
  2. To safely perform index-based operations on a CopyOnWriteArrayList that can be concurrently modified or shrunk, snapshot the list to an array (e.g., using toArray()) and perform all index-based accesses on the array snapshot to prevent IndexOutOfBoundsException.

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<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 RecordingScheduler(FakeApiClock clock) {
super(CORE_POOL_SIZE);
this.clock = clock;
}

// List<Runnable> shutdownNow()
when(mock.shutdownNow())
.then(
new Answer<List<Runnable>>() {
@Override
public List<Runnable> 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<java.time.Duration> 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.
*
* <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.
*/
@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<Integer>() {
@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<Duration> getSleepDurations() {
return sleepDurations;
}

return mock;
/** Returns the total number of tasks scheduled via {@link #schedule(Runnable, long, TimeUnit)}. */
public int getIterationsCount() {
return iterationsCount.get();
}
}
7 changes: 6 additions & 1 deletion .../gax-java/gax/src/test/java/com/google/api/gax/retrying/AbstractRetryingExecutorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,7 @@
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoMoreInteractions;

import com.google.api.core.ApiClock;
import com.google.api.core.ApiFuture;
import com.google.api.core.NanoClock;
import com.google.api.gax.retrying.FailingCallable.CustomException;
Expand Down Expand Up @@ -102,6 +103,10 @@ protected abstract RetryingExecutorWithContext<String> getExecutor(
protected abstract RetryAlgorithm<String> getAlgorithm(
RetrySettings retrySettings, int apocalypseCountDown, RuntimeException apocalypseException);

protected ApiClock getClock() {
return NanoClock.getDefaultClock();
}

protected <T> void busyWaitForInitialResult(RetryingFuture<T> future, java.time.Duration timeout)
throws TimeoutException {
Stopwatch watch = Stopwatch.createStarted();
Expand Down Expand Up @@ -356,7 +361,7 @@ void testPollExceptionByPollAlgorithm(boolean withCustomRetrySettings) throws Ex
useContextRetrySettings
? withCustomRetrySettings ? FAILING_RETRY_SETTINGS : FAST_RETRY_SETTINGS
: retrySettings,
NanoClock.getDefaultClock()));
getClock()));

RetryingExecutorWithContext<String> executor = getExecutor(retryAlgorithm);
FailingCallable callable = new FailingCallable(6, "request", "FAILURE", tracer);
Expand Down
27 changes: 24 additions & 3 deletions ...va/gax-java/gax/src/test/java/com/google/api/gax/retrying/DirectRetryingExecutorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,20 +29,41 @@
*/
package com.google.api.gax.retrying;

import com.google.api.core.CurrentMillisClock;
import com.google.api.core.ApiClock;
import com.google.api.gax.core.FakeApiClock;
import java.time.Duration;

class DirectRetryingExecutorTest extends AbstractRetryingExecutorTest {

// ExponentialRetryAlgorithm and ExponentialPollAlgorithm only compute relative elapsed time
// deltas against firstAttemptStartTimeNanos (clock.nanoTime() - firstAttemptStartTimeNanos)
// and never treat 0L as a special sentinel value.
private final FakeApiClock clock = new FakeApiClock(0L);

@Override
protected ApiClock getClock() {
return clock;
}

@Override
protected RetryingExecutorWithContext<String> getExecutor(RetryAlgorithm<String> retryAlgorithm) {
return new DirectRetryingExecutor<>(retryAlgorithm);
return new DirectRetryingExecutor<String>(retryAlgorithm) {
// Advance the fake clock instead of sleeping the current thread so that retry delays
// are simulated deterministically without wall-clock pauses on slow CI VMs.
@Override
protected void sleep(Duration delay) {
if (!delay.isZero() && !delay.isNegative()) {
clock.incrementNanoTime(delay.toNanos());
}
}
};
}

@Override
protected RetryAlgorithm<String> getAlgorithm(
RetrySettings retrySettings, int apocalypseCountDown, RuntimeException apocalypseException) {
return new RetryAlgorithm<>(
new TestResultRetryAlgorithm<String>(apocalypseCountDown, apocalypseException),
new ExponentialRetryAlgorithm(retrySettings, CurrentMillisClock.getDefaultClock()));
new ExponentialRetryAlgorithm(retrySettings, clock));
}
}
33 changes: 29 additions & 4 deletions ...gax-java/gax/src/test/java/com/google/api/gax/retrying/ScheduledRetryingExecutorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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(
Expand Down Expand Up @@ -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,
Expand Down
Loading