CooperatingFuture extends CompletableFuture<T> { @VisibleForTesting Time staggerTimeout(final Time gap) { long currentTimeMillis = System.currentTimeMillis(); long prevTimeMillis, nextTimeMillis; do { prevTimeMillis = staggerTimeMillis.get(); nextTimeMillis = Math.max(prevTimeMillis + gap.toMillis(), currentTimeMillis); } while (!staggerTimeMillis.compareAndSet(prevTimeMillis, nextTimeMillis)); return Time.millis(nextTimeMillis - currentTimeMillis); } CooperatingFuture(final String requestKey, final Config config); T call(final IOCall<T> request); T cooperate(final IOCall<T> request); @VisibleForTesting String getRequestKey(); @VisibleForTesting int getThreadCount(); @Override String toString();  }
@Test public void downloadTimeoutsAreStaggered() { CooperatingFuture<String> cooperatingFuture = new CooperatingFuture<>("testKey", config); Random random = new Random(); long[] downloadTimeMillis = new long[10]; long expectedGap = 200; downloadTimeMillis[0] = System.currentTimeMillis(); for (int i = 1; i < downloadTimeMillis.length; i++) { LockSupport.parkNanos(Time.millis(random.nextInt((int) expectedGap)).toNanos()); LockSupport.parkNanos(cooperatingFuture.staggerTimeout(Time.millis(expectedGap)).toNanos()); downloadTimeMillis[i] = System.currentTimeMillis(); } for (int i = 1; i < downloadTimeMillis.length; i++) { long actualGap = downloadTimeMillis[i] - downloadTimeMillis[i - 1]; assertThat(actualGap, allOf(greaterThanOrEqualTo(expectedGap - 10), lessThanOrEqualTo(expectedGap + 10))); } }