graphql-java
diff --git a/README.md b/README.md
@@ -633,6 +633,42 @@ Do not assume that a single call to `dispatch()` results in a single call to `Ba
 
 This code is inspired from the scheduling code in the [reference JS implementation](https://github.com/graphql/dataloader#batch-scheduling)
 
+### Scheduling DataLoader promise completion
+
+Another step in the batch loading process is the completion of the futures that have been promised earlier via
+`Dataloader.load()`. This can also be scheduled on another thread if need be, allowing the dispatch call 
+to return quicker.
+
+You might want to schedule the completion of values on another thread if there is following work t do such as :
+
+```java
+var cfStage1 = dataLoader.load(key);
+var cfStage2 = cfStage1.thenApply(v -> doSomethingSlow(v));
+//...
+var dispatchCF = dataLoader.dispatch();
+```     
+
+In the above example the `.doSomethingSlow(v)` call will happen inside the completion
+of the `cfStage1` code path since `CompletableFuture` by design eagerly runs dependent chained methods.
+
+Perhaps you want this completion to happen more asynchronously so that the `.dispatch()` returns more
+quickly and is not bound to the `.doSomethingSlow(v)` call.
+
+By default, the dispatch completion is done on the current thread in a synchronous manner, which will include
+any extra `CompletableFuture` chained methods. If multiple keys were loaded, the dispatch completion is done
+sequentially.
+
+This is an example of running the completion step in an asynchronous manner :
+
+```java
+
+@Override
+public <K> CompletionStage<Void> scheduleCompletion(List<Runnable> completeValueRunnables, List<K> keys, BatchLoaderEnvironment environment) {
+    return CompletableFuture.allOf(completeValueRunnables.stream()
+            .map(CompletableFuture::runAsync).toArray(CompletableFuture[]::new));
+}
+```
+
 ## Scheduled Registry Dispatching
 
 `ScheduledDataLoaderRegistry` is a registry that allows for dispatching to be done on a schedule. It contains a

diff --git a/src/main/java/org/dataloader/DataLoaderHelper.java b/src/main/java/org/dataloader/DataLoaderHelper.java
@@ -28,11 +28,11 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
 import java.util.concurrent.CompletionStage;
+import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.atomic.AtomicReference;
 
 import static java.util.Collections.emptyList;
 import static java.util.Collections.singletonList;
-import static java.util.concurrent.CompletableFuture.allOf;
 import static java.util.concurrent.CompletableFuture.completedFuture;
 import static java.util.stream.Collectors.toList;
 import static org.dataloader.impl.Assertions.assertState;
@@ -244,7 +244,7 @@ DispatchResult<V> dispatch() {
         int queueSize = loaderQueueEntryHead.queueSize;
         // we copy the pre-loaded set of futures ready for dispatch
         Object[] keysArray = new Object[queueSize];
-        CompletableFuture[] queuedFuturesArray = new CompletableFuture[queueSize];
+        CompletableFuture<V>[] queuedFuturesArray = new CompletableFuture[queueSize];
         Object[] callContextsArray = new Object[queueSize];
         int index = queueSize - 1;
         while (loaderQueueEntryHead != null) {
@@ -313,54 +313,59 @@ private CompletableFuture<List<V>> sliceIntoBatchesOfBatches(List<K> keys, List<
         }
         //
         // now reassemble all the futures into one that is the complete set of results
-        return allOf(allBatches.toArray(new CompletableFuture[0]))
-                .thenApply(v -> allBatches.stream()
-                        .map(CompletableFuture::join)
-                        .flatMap(Collection::stream)
-                        .collect(toList()));
+        return CompletableFutureKit.allOfFlatMap(allBatches);
     }
 
     @SuppressWarnings("unchecked")
-    private CompletableFuture<List<V>> dispatchQueueBatch(List<K> keys, List<Object> callContexts, List<CompletableFuture<V>> queuedFutures) {
-        stats.incrementBatchLoadCountBy(keys.size(), new IncrementBatchLoadCountByStatisticsContext<>(keys, callContexts));
-        CompletableFuture<List<V>> batchLoad = invokeLoader(keys, callContexts, queuedFutures, loaderOptions.cachingEnabled());
+    private CompletableFuture<List<V>> dispatchQueueBatch(List<K> keys, List<Object> keyContexts, List<CompletableFuture<V>> queuedFutures) {
+        stats.incrementBatchLoadCountBy(keys.size(), new IncrementBatchLoadCountByStatisticsContext<>(keys, keyContexts));
+
+        BatchLoaderEnvironment environment = mkBatchLoaderEnv(keys, keyContexts);
+
+        CompletableFuture<List<V>> batchLoad = invokeLoader(environment, keys, keyContexts, queuedFutures, loaderOptions.cachingEnabled());
         return batchLoad
-                .thenApply(values -> {
+                .thenCompose(values -> {
                     assertResultSize(keys, values);
                     if (isPublisher() || isMappedPublisher()) {
                         // We have already completed the queued futures by the time the overall batchLoad future has completed.
-                        return values;
+                        return CompletableFutureKit.success(values);
                     }
 
-                    List<K> clearCacheKeys = new ArrayList<>();
+                    Collection<K> clearCacheKeys = new ConcurrentLinkedQueue<>();
+                    List<Runnable> completeValueRunnables = new ArrayList<>();
                     for (int idx = 0; idx < queuedFutures.size(); idx++) {
                         K key = keys.get(idx);
                         V value = values.get(idx);
-                        Object callContext = callContexts.get(idx);
+                        Object callContext = keyContexts.get(idx);
                         CompletableFuture<V> future = queuedFutures.get(idx);
-                        if (value instanceof Throwable) {
-                            stats.incrementLoadErrorCount(new IncrementLoadErrorCountStatisticsContext<>(key, callContext));
-                            future.completeExceptionally((Throwable) value);
-                            clearCacheKeys.add(keys.get(idx));
-                        } else if (value instanceof Try) {
-                            // we allow the batch loader to return a Try so we can better represent a computation
-                            // that might have worked or not.
-                            Try<V> tryValue = (Try<V>) value;
-                            if (tryValue.isSuccess()) {
-                                future.complete(tryValue.get());
-                            } else {
+                        Runnable completeValueRunnable = () -> {
+                            if (value instanceof Throwable) {
                                 stats.incrementLoadErrorCount(new IncrementLoadErrorCountStatisticsContext<>(key, callContext));
-                                future.completeExceptionally(tryValue.getThrowable());
-                                clearCacheKeys.add(keys.get(idx));
+                                future.completeExceptionally((Throwable) value);
+                                clearCacheKeys.add(key);
+                            } else if (value instanceof Try) {
+                                // we allow the batch loader to return a Try so we can better represent a computation
+                                // that might have worked or not.
+                                Try<V> tryValue = (Try<V>) value;
+                                if (tryValue.isSuccess()) {
+                                    future.complete(tryValue.get());
+                                } else {
+                                    stats.incrementLoadErrorCount(new IncrementLoadErrorCountStatisticsContext<>(key, callContext));
+                                    future.completeExceptionally(tryValue.getThrowable());
+                                    clearCacheKeys.add(key);
+                                }
+                            } else {
+                                future.complete(value);
                             }
-                        } else {
-                            future.complete(value);
-                        }
+                        };
+                        completeValueRunnables.add(completeValueRunnable);
                     }
-                    possiblyClearCacheEntriesOnExceptions(clearCacheKeys);
-                    return values;
+                    return scheduleCompletion(environment, keys, values, completeValueRunnables).thenApply(ignored -> {
+                        possiblyClearCacheEntriesOnExceptions(clearCacheKeys);
+                        return values;
+                    });
                 }).exceptionally(ex -> {
-                    stats.incrementBatchLoadExceptionCount(new IncrementBatchLoadExceptionCountStatisticsContext<>(keys, callContexts));
+                    stats.incrementBatchLoadExceptionCount(new IncrementBatchLoadExceptionCountStatisticsContext<>(keys, keyContexts));
                     if (ex instanceof CompletionException) {
                         ex = ex.getCause();
                     }
@@ -375,12 +380,32 @@ private CompletableFuture<List<V>> dispatchQueueBatch(List<K> keys, List<Object>
                 });
     }
 
+    private CompletableFuture<List<V>> scheduleCompletion(BatchLoaderEnvironment environment, List<K> keys, List<V> values, List<Runnable> completeValueRunnables) {
+        BatchLoaderScheduler batchLoaderScheduler = loaderOptions.getBatchLoaderScheduler();
+        CompletionStage<?> scheduledCompletion;
+        if (batchLoaderScheduler != null) {
+            scheduledCompletion = batchLoaderScheduler
+                    .scheduleCompletion(completeValueRunnables, keys, environment);
+        } else {
+            scheduledCompletion = CompletableFutureKit.runAll(completeValueRunnables);
+        }
+        return scheduledCompletion
+                .thenApply(ignored -> values)
+                .toCompletableFuture();
+    }
+
+    private BatchLoaderEnvironment mkBatchLoaderEnv(List<K> keys, List<Object> keyContexts) {
+        Object context = loaderOptions.getBatchLoaderContextProvider().getContext();
+        return BatchLoaderEnvironment.newBatchLoaderEnvironment()
+                .context(context).keyContexts(keys, keyContexts).build();
+    }
+
 
     private void assertResultSize(List<K> keys, List<V> values) {
         assertState(keys.size() == values.size(), () -> "The size of the promised values MUST be the same size as the key list");
     }
 
-    private void possiblyClearCacheEntriesOnExceptions(List<K> keys) {
+    private void possiblyClearCacheEntriesOnExceptions(Collection<K> keys) {
         if (keys.isEmpty()) {
             return;
         }
@@ -396,15 +421,17 @@ private void possiblyClearCacheEntriesOnExceptions(List<K> keys) {
     CompletableFuture<V> invokeLoaderImmediately(K key, Object keyContext, boolean cachingEnabled) {
         List<K> keys = singletonList(key);
         List<Object> keyContexts = singletonList(keyContext);
+        BatchLoaderEnvironment environment = mkBatchLoaderEnv(keys, keyContexts);
+
         List<CompletableFuture<V>> queuedFutures = singletonList(new CompletableFuture<>());
-        return invokeLoader(keys, keyContexts, queuedFutures, cachingEnabled)
+        return invokeLoader(environment, keys, keyContexts, queuedFutures, cachingEnabled)
                 .thenApply(list -> list.get(0))
                 .toCompletableFuture();
     }
 
-    CompletableFuture<List<V>> invokeLoader(List<K> keys, List<Object> keyContexts, List<CompletableFuture<V>> queuedFutures, boolean cachingEnabled) {
+    CompletableFuture<List<V>> invokeLoader(BatchLoaderEnvironment environment, List<K> keys, List<Object> keyContexts, List<CompletableFuture<V>> queuedFutures, boolean cachingEnabled) {
         if (!cachingEnabled) {
-            return invokeLoader(keys, keyContexts, queuedFutures);
+            return invokeLoader(environment, keys, keyContexts, queuedFutures);
         }
         CompletableFuture<List<Try<V>>> cacheCallCF = getFromValueCache(keys);
         return cacheCallCF.thenCompose(cachedValues -> {
@@ -453,7 +480,8 @@ CompletableFuture<List<V>> invokeLoader(List<K> keys, List<Object> keyContexts,
                 // we missed some keys from cache, so send them to the batch loader
                 // and then fill in their values
                 //
-                CompletableFuture<List<V>> batchLoad = invokeLoader(missedKeys, missedKeyContexts, missedQueuedFutures);
+                BatchLoaderEnvironment missedEnvironment = mkBatchLoaderEnv(missedKeys, missedKeyContexts);
+                CompletableFuture<List<V>> batchLoad = invokeLoader(missedEnvironment, missedKeys, missedKeyContexts, missedQueuedFutures);
                 return batchLoad.thenCompose(missedValues -> {
                     assertResultSize(missedKeys, missedValues);
 
@@ -472,11 +500,7 @@ CompletableFuture<List<V>> invokeLoader(List<K> keys, List<Object> keyContexts,
         });
     }
 
-    CompletableFuture<List<V>> invokeLoader(List<K> keys, List<Object> keyContexts, List<CompletableFuture<V>> queuedFutures) {
-        Object context = loaderOptions.getBatchLoaderContextProvider().getContext();
-        BatchLoaderEnvironment environment = BatchLoaderEnvironment.newBatchLoaderEnvironment()
-                .context(context).keyContexts(keys, keyContexts).build();
-
+    CompletableFuture<List<V>> invokeLoader(BatchLoaderEnvironment environment, List<K> keys, List<Object> keyContexts, List<CompletableFuture<V>> queuedFutures) {
         DataLoaderInstrumentationContext<List<?>> instrCtx = ctxOrNoopCtx(instrumentation().beginBatchLoader(dataLoader, keys, environment));
 
         CompletableFuture<List<V>> batchLoad;

diff --git a/src/main/java/org/dataloader/impl/CompletableFutureKit.java b/src/main/java/org/dataloader/impl/CompletableFutureKit.java
@@ -2,6 +2,7 @@
 
 import org.dataloader.annotations.Internal;
 
+import java.util.Collection;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
@@ -22,6 +23,12 @@ public static <V> CompletableFuture<V> failedFuture(Exception e) {
         return future;
     }
 
+    public static <V> CompletableFuture<V> success(V v) {
+        CompletableFuture<V> future = new CompletableFuture<>();
+        future.complete(v);
+        return future;
+    }
+
     public static <V> Throwable cause(CompletableFuture<V> completableFuture) {
         if (!completableFuture.isCompletedExceptionally()) {
             return null;
@@ -67,4 +74,41 @@ public static <K, V> CompletableFuture<Map<K, V>> allOf(Map<K, CompletableFuture
                         )
                 );
     }
+
+    public static <T> CompletableFuture<List<T>> allOfFlatMap(List<CompletableFuture<List<T>>> cfs) {
+
+        return CompletableFuture.allOf(cfs.toArray(new CompletableFuture[0]))
+                .thenApply(v -> cfs.stream()
+                        .map(CompletableFuture::join)
+                        .flatMap(Collection::stream)
+                        .collect(toList()));
+    }
+
+    /**
+     * Runs the given {@link Runnable} synchronously on the current thread, returning a
+     * {@link CompletableFuture} that is completed normally or exceptionally based on the outcome.
+     *
+     * @param runnable the task to execute
+     * @return a completed future, or a failed future if the runnable throws
+     */
+    public static CompletableFuture<Void> run(Runnable runnable) {
+        try {
+            runnable.run();
+            return CompletableFutureKit.success(null);
+        } catch (Exception e) {
+            return CompletableFutureKit.failedFuture(e);
+        }
+    }
+
+    /**
+     * Runs all the {@link Runnable} from the list synchronously on the current thread, returning a
+     * {@link CompletableFuture} that is completed normally or exceptionally based on the outcome.
+     *
+     * @param runnables the list of tasks to execute
+     * @return a completed future, or a failed future if any of the tasks throws
+     */
+    public static CompletableFuture<Void> runAll(List<Runnable> runnables) {
+        return CompletableFuture.allOf(runnables.stream().map(CompletableFutureKit::run).toArray(CompletableFuture[]::new));
+    }
+
 }
diff --git a/src/main/java/org/dataloader/scheduler/BatchLoaderScheduler.java b/src/main/java/org/dataloader/scheduler/BatchLoaderScheduler.java
@@ -2,11 +2,12 @@
 
 import org.dataloader.BatchLoader;
 import org.dataloader.BatchLoaderEnvironment;
+import org.dataloader.BatchPublisher;
 import org.dataloader.DataLoader;
 import org.dataloader.DataLoaderOptions;
 import org.dataloader.MappedBatchLoader;
 import org.dataloader.MappedBatchPublisher;
-import org.dataloader.BatchPublisher;
+import org.dataloader.impl.CompletableFutureKit;
 
 import java.util.List;
 import java.util.Map;
@@ -31,6 +32,7 @@ public interface BatchLoaderScheduler {
      * @param <V> the value type
      */
     interface ScheduledBatchLoaderCall<V> {
+
         CompletionStage<List<V>> invoke();
     }
 
@@ -41,13 +43,15 @@ interface ScheduledBatchLoaderCall<V> {
      * @param <V> the value type
      */
     interface ScheduledMappedBatchLoaderCall<K, V> {
+
         CompletionStage<Map<K, V>> invoke();
     }
 
     /**
      * This represents a callback that will invoke a {@link BatchPublisher} or {@link MappedBatchPublisher} function under the covers
      */
     interface ScheduledBatchPublisherCall {
+
         void invoke();
     }
 
@@ -92,4 +96,40 @@ interface ScheduledBatchPublisherCall {
      * @param <K>           the key type
      */
     <K> void scheduleBatchPublisher(ScheduledBatchPublisherCall scheduledCall, List<K> keys, BatchLoaderEnvironment environment);
+
+    /**
+     * This is called to schedule the "completion" of the {@link java.util.concurrent.CompletableFuture}s in the {@link org.dataloader.DataLoader} that map
+     * to values that have come back from the batch loader call.
+     * <p>
+     * You might want to schedule the completion of values on another thread if there following chained work such as :
+     * <pre>
+     * {@code
+     *   var cfStage1 = dataLoader.load(key);
+     *   var cfStage2 = cfStage1.thenApply(v -> doSomethingSlow(v));
+     *   //...
+     *   var dispatchCF = dataLoader.dispatch();
+     * }
+     * </pre>
+     * <p>
+     * In the above example the `.doSomethingSlow(v)` call will happen inside the completion
+     * of the dataloader code path when it tries to complete `cfStage1` since {@link java.util.concurrent.CompletableFuture}
+     * by design runs chained dependent methods eagerly.
+     * <p>
+     * Perhaps you want this tp happen more asynchronously so that the `.dispatch()` returns more
+     * quickly and is not bound to the `.doSomethingSlow(v)` call.
+     * <p>
+     * By default, the dispatch completion is done on the current thread in a synchronous manner, which will include
+     * any extra {@link java.util.concurrent.CompletableFuture} dependent chained methods.
+     *
+     * @param completeValueRunnables these are the runnable tasks that the {@link DataLoader} engine code needs to be run
+     * @param keys          this is the list of keys that will be passed to the {@link BatchPublisher}.
+     *                      This is provided only for informative reasons and, you can't change the keys that are used
+     * @param environment   this is the {@link BatchLoaderEnvironment} in place,
+     *
+     * @return a {@link CompletionStage} representing this work is being scheduled
+     */
+    default <K> CompletionStage<Void> scheduleCompletion(List<Runnable> completeValueRunnables, List<K> keys, BatchLoaderEnvironment environment) {
+        return CompletableFutureKit.runAll(completeValueRunnables);
+    }
+
 }