Executors should not be used for getStage calls

Author: jhalterman

src/main/java/net/jodah/failsafe/FailsafeExecutor.java

@@ -200,7 +200,7 @@ public <T extends R> CompletableFuture<T> getAsyncExecution(AsyncRunnable<T> run
    * @throws RejectedExecutionException if the {@code supplier} cannot be scheduled for execution
    */
   public <T extends R> CompletableFuture<T> getStageAsync(CheckedSupplier<? extends CompletionStage<T>> supplier) {
-    return callAsync(future -> getPromiseOfStage(toCtxSupplier(supplier), future, executor), false);
+    return callAsync(future -> getPromiseOfStage(toCtxSupplier(supplier), future), false);
   }
 
   /**
@@ -222,7 +222,7 @@ public <T extends R> CompletableFuture<T> getStageAsync(CheckedSupplier<? extend
    */
   public <T extends R> CompletableFuture<T> getStageAsync(
     ContextualSupplier<T, ? extends CompletionStage<T>> supplier) {
-    return callAsync(future -> getPromiseOfStage(supplier, future, executor), false);
+    return callAsync(future -> getPromiseOfStage(supplier, future), false);
   }
 
   /**
@@ -250,7 +250,7 @@ public <T extends R> CompletableFuture<T> getStageAsync(
    */
   public <T extends R> CompletableFuture<T> getStageAsyncExecution(
     AsyncSupplier<T, ? extends CompletionStage<T>> supplier) {
-    return callAsync(future -> getPromiseOfStageExecution(supplier, future, executor), true);
+    return callAsync(future -> getPromiseOfStageExecution(supplier, future), true);
   }
 
   /**
@@ -408,6 +408,9 @@ public FailsafeExecutor<R> with(ExecutorService executorService) {
    * Configures the {@code executor} to use as a wrapper around executions. The {@code executor} is responsible for
    * propagating executions. Executions that normally return a result, such as {@link #get(CheckedSupplier)} will return
    * {@code null} since the {@link Executor} interface does not support results.
+   * <p>The {@code executor} will not be used for {@link #getStageAsync(CheckedSupplier) getStageAsync} or {@link
+   * #getStageAsyncExecution(AsyncSupplier) getStageAsyncExecution} calls since those require a returned result.
+   * </p>
    *
    * @throws NullPointerException if {@code executor} is null
    */

src/main/java/net/jodah/failsafe/Functions.java

@@ -121,18 +121,18 @@ public synchronized CompletableFuture<ExecutionResult<R>> apply(AsyncExecutionIn
    * supplier}, records the result and returns a promise containing the result.
    *
    * @param <R> result type
+   * @throws UnsupportedOperationException when using
    */
   @SuppressWarnings("unchecked")
   static <R> Function<AsyncExecutionInternal<R>, CompletableFuture<ExecutionResult<R>>> getPromiseOfStage(
-    ContextualSupplier<R, ? extends CompletionStage<? extends R>> supplier, FailsafeFuture<R> future,
-    Executor executor) {
+    ContextualSupplier<R, ? extends CompletionStage<? extends R>> supplier, FailsafeFuture<R> future) {
 
     Assert.notNull(supplier, "supplier");
     return execution -> {
       CompletableFuture<ExecutionResult<R>> promise = new CompletableFuture<>();
       try {
         execution.preExecute();
-        CompletionStage<? extends R> stage = withExecutor(supplier, executor).get(execution);
+        CompletionStage<? extends R> stage = supplier.get(execution);
 
         // Propagate outer cancellations to the stage
         if (stage instanceof Future)
@@ -163,15 +163,15 @@ static <R> Function<AsyncExecutionInternal<R>, CompletableFuture<ExecutionResult
    */
   @SuppressWarnings("unchecked")
   static <R> Function<AsyncExecutionInternal<R>, CompletableFuture<ExecutionResult<R>>> getPromiseOfStageExecution(
-    AsyncSupplier<R, ? extends CompletionStage<? extends R>> supplier, FailsafeFuture<R> future, Executor executor) {
+    AsyncSupplier<R, ? extends CompletionStage<? extends R>> supplier, FailsafeFuture<R> future) {
 
     Assert.notNull(supplier, "supplier");
     Semaphore asyncFutureLock = new Semaphore(1);
     return execution -> {
       try {
         execution.preExecute();
         asyncFutureLock.acquire();
-        CompletionStage<? extends R> stage = withExecutor(supplier, executor).get(execution);
+        CompletionStage<? extends R> stage = supplier.get(execution);
 
         // Propagate outer cancellations to the stage
         if (stage instanceof Future)
@@ -305,19 +305,6 @@ static <R> AsyncRunnable<R> withExecutor(AsyncRunnable<R> runnable, Executor exe
     };
   }
 
-  static <R, T> AsyncSupplier<R, T> withExecutor(AsyncSupplier<R, T> supplier, Executor executor) {
-    return executor == null ? supplier : exec -> {
-      executor.execute(() -> {
-        try {
-          supplier.get(exec);
-        } catch (Throwable e) {
-          handleExecutorThrowablen(e);
-        }
-      });
-      return null;
-    };
-  }
-
   private static void handleExecutorThrowablen(Throwable e) {
     if (e instanceof RuntimeException)
       throw (RuntimeException) e;