Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
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
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
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<>();
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();
}
}
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