FutureUtil { public static <T> Future<T> getSpeculativeFuture( final Future<T> originalFuture, final Function0<Future<T>> functionBackupFuture, long waitTimeMillis, final String statsPrefix) { return getSpeculativeFuture( originalFuture, functionBackupFuture, waitTimeMillis, statsPrefix, BackupFutureRetryPolicy.UN_CONDITIONAL_RETRY_POLICY); }  static Future<T> getSpeculativeFuture(
      final Future<T> originalFuture,
      final Function0<Future<T>> functionBackupFuture,
      long waitTimeMillis,
      final String statsPrefix); static Future<T> getSpeculativeFuture(
      final Future<T> originalFuture,
      final Function0<Future<T>> functionBackupFuture,
      long waitTimeMillis,
      final String statsPrefix,
      final BackupFutureRetryPolicy retryPolicy); static Future<Try<T>> lifeToTry(Future<T> future);  }
@Test public void testNoBackupFutureExecutedSuccess() { Future<Integer> originalFuture = getFuture(50, 1, true); Function0<Future<Integer>> backupFunctionFuture = getBackupFutureFunctionWithSuccess(500, 2); Future<Integer> speculativeFuture = FutureUtil.getSpeculativeFuture( originalFuture, backupFunctionFuture, 100, STATS_PREFIX); assertEquals(1, speculativeFuture.get().intValue()); assertFalse(isBackupFutureInvoked); checkStats(0, 0, 0); }
@Test public void testNoBackupFutureExecutedEarlyFailureForTimeBasedPolicy() { Future<Integer> originalFuture = getFuture(50, 1, false); Function0<Future<Integer>> backupFunctionFuture = getBackupFutureFunctionWithSuccess(500, 2); Future<Integer> speculativeFuture = FutureUtil.getSpeculativeFuture( originalFuture, backupFunctionFuture, 100, STATS_PREFIX, BackupFutureRetryPolicy.TIMEOUT_RETRY_POLICY); Exception e = null; try { speculativeFuture.get(); } catch (RuntimeException re) { e = re; } assertNotNull(e); assertEquals(EXCEPTION_MSG, e.getMessage()); assertFalse(isBackupFutureInvoked); checkStats(0, 0, 0); }
@Test public void testBackupFutureExecutedEarlyFailure() { Future<Integer> originalFuture = getFuture(50, 1, false); Function0<Future<Integer>> backupFunctionFuture = getBackupFutureFunctionWithSuccess(500, 2); Future<Integer> speculativeFuture = FutureUtil.getSpeculativeFuture( originalFuture, backupFunctionFuture, 100, STATS_PREFIX, BackupFutureRetryPolicy.UN_CONDITIONAL_RETRY_POLICY); speculativeFuture.get(); assertTrue(isBackupFutureInvoked); checkStats(1, 0, 1); }
@Test public void testBothFutureSuccessAndOriginalFinishesFirst() { Future<Integer> originalFuture = getFuture(200, 1, true); Function0<Future<Integer>> backupFunctionFuture = getBackupFutureFunctionWithSuccess(1000, 2); Future<Integer> speculativeFuture = FutureUtil.getSpeculativeFuture( originalFuture, backupFunctionFuture, 100, STATS_PREFIX); assertEquals(1, speculativeFuture.get().intValue()); assertTrue(isBackupFutureInvoked); checkStats(1, 1, 0); }
@Test public void testBothFutureSuccessAndBackupFinishesFirst() { Future<Integer> originalFuture = getFuture(1000, 1, true); Function0<Future<Integer>> backupFunctionFuture = getBackupFutureFunctionWithSuccess(200, 2); Future<Integer> speculativeFuture = FutureUtil.getSpeculativeFuture( originalFuture, backupFunctionFuture, 100, STATS_PREFIX); int value = speculativeFuture.get().intValue(); assertTrue(isBackupFutureInvoked); assertEquals(2, value); checkStats(1, 0, 1); }
@Test public void testOriginalFailsAndBackupFinishesSecond() { Future<Integer> originalFuture = getFuture(300, 1, false); Function0<Future<Integer>> backupFunctionFuture = getBackupFutureFunctionWithSuccess(800, 2); Future<Integer> speculativeFuture = FutureUtil.getSpeculativeFuture( originalFuture, backupFunctionFuture, 100, STATS_PREFIX); assertEquals(2, speculativeFuture.get().intValue()); assertTrue(isBackupFutureInvoked); checkStats(1, 0, 1); }
@Test public void testBackupFailsAndOriginalFinishesSecond() { Future<Integer> originalFuture = getFuture(1000, 1, true); Function0<Future<Integer>> backupFunctionFuture = getBackupFutureFunctionWithFailure(400, 2); Future<Integer> speculativeFuture = FutureUtil.getSpeculativeFuture( originalFuture, backupFunctionFuture, 100, STATS_PREFIX); assertEquals(1, speculativeFuture.get().intValue()); assertTrue(isBackupFutureInvoked); checkStats(1, 1, 0); }
@Test public void testOriginalFailsAndBackupFinishesFirst() { Future<Integer> originalFuture = getFuture(500, 1, false); Function0<Future<Integer>> backupFunctionFuture = getBackupFutureFunctionWithSuccess(10, 2); Future<Integer> speculativeFuture = FutureUtil.getSpeculativeFuture( originalFuture, backupFunctionFuture, 100, STATS_PREFIX); assertEquals(2, speculativeFuture.get().intValue()); assertTrue(isBackupFutureInvoked); checkStats(1, 0, 1); }
@Test public void testBackupFailsAndOriginalFinishesFirst() { Future<Integer> originalFuture = getFuture(300, 1, true); Function0<Future<Integer>> backupFunctionFuture = getBackupFutureFunctionWithFailure(800, 2); Future<Integer> speculativeFuture = FutureUtil.getSpeculativeFuture( originalFuture, backupFunctionFuture, 100, STATS_PREFIX); assertEquals(1, speculativeFuture.get().intValue()); assertTrue(isBackupFutureInvoked); checkStats(1, 1, 0); }
@Test public void testBothFutureFailure() { Future<Integer> originalFuture = getFuture(200, 1, false); Function0<Future<Integer>> backupFunctionFuture = getBackupFutureFunctionWithFailure(200, 2); Future<Integer> speculativeFuture = FutureUtil.getSpeculativeFuture( originalFuture, backupFunctionFuture, 100, STATS_PREFIX); Exception e = null; try { speculativeFuture.get(); } catch (RuntimeException re) { e = re; } assertNotNull(e); assertEquals(EXCEPTION_MSG, e.getMessage()); assertTrue(isBackupFutureInvoked); checkStats(1, 0, 0); }