diff --git a/README.md b/README.md
index 1d2b4e5..ab80f6e 100644
--- a/README.md
+++ b/README.md
@@ -168,16 +168,17 @@ That said, with key caching turn on (the default), it will still be more efficie
 ### Calling the batch loader function with call context environment
 
 Often there is a need to call the batch loader function with some sort of call context environment, such as the calling users security
-credentials or the database connection parameters.  You can do this by implementing a 
-`org.dataloader.BatchLoaderEnvironmentProvider` and using one of the `xxxWithContext` batch loading interfaces
-such as `org.dataloader.BatchLoaderWithContext`.
+credentials or the database connection parameters.  
 
-```java
-        BatchLoaderEnvironment batchLoaderEnvironment = BatchLoaderEnvironment.newBatchLoaderEnvironment()
-                .context(SecurityCtx.getCallingUserCtx()).build();
+You can do this by implementing a `org.dataloader.BatchLoaderContextProvider` and using one of 
+the batch loading interfaces such as `org.dataloader.BatchLoaderWithContext`.
+
+It will be given a `org.dataloader.BatchLoaderEnvironment` parameter and it can then ask it
+for the context object.
 
+```java
         DataLoaderOptions options = DataLoaderOptions.newOptions()
-                .setBatchLoaderEnvironmentProvider(() -> batchLoaderEnvironment);
+                .setBatchLoaderEnvironmentProvider(() -> SecurityCtx.getCallingUserCtx());
 
         BatchLoaderWithContext<String, String> batchLoader = new BatchLoaderWithContext<String, String>() {
             @Override
diff --git a/src/main/java/org/dataloader/BatchLoaderContextProvider.java b/src/main/java/org/dataloader/BatchLoaderContextProvider.java
new file mode 100644
index 0000000..b32c3f0
--- /dev/null
+++ b/src/main/java/org/dataloader/BatchLoaderContextProvider.java
@@ -0,0 +1,13 @@
+package org.dataloader;
+
+/**
+ * A BatchLoaderContextProvider is used by the {@link org.dataloader.DataLoader} code to
+ * provide overall calling context to the {@link org.dataloader.BatchLoader} call.  A common use
+ * case is for propagating user security credentials or database connection parameters for example.
+ */
+public interface BatchLoaderContextProvider {
+    /**
+     * @return a context object that may be needed in batch load calls
+     */
+    Object getContext();
+}
\ No newline at end of file
diff --git a/src/main/java/org/dataloader/BatchLoaderEnvironment.java b/src/main/java/org/dataloader/BatchLoaderEnvironment.java
index c6d0126..8120673 100644
--- a/src/main/java/org/dataloader/BatchLoaderEnvironment.java
+++ b/src/main/java/org/dataloader/BatchLoaderEnvironment.java
@@ -1,5 +1,10 @@
 package org.dataloader;
 
+import java.util.Collections;
+import java.util.Map;
+
+import static org.dataloader.impl.Assertions.nonNull;
+
 /**
  * This object is passed to a batch loader as calling context.  It could contain security credentials
  * of the calling users for example or database parameters that allow the data layer call to succeed.
@@ -7,22 +12,43 @@
 public class BatchLoaderEnvironment {
 
     private final Object context;
+    private final Map<Object, Object> keyContexts;
 
-    private BatchLoaderEnvironment(Object context) {
+    private BatchLoaderEnvironment(Object context, Map<Object, Object> keyContexts) {
         this.context = context;
+        this.keyContexts = keyContexts;
     }
 
+    /**
+     * Returns the overall context object provided by {@link org.dataloader.BatchLoaderContextProvider}
+     *
+     * @param <T> the type you would like the object to be
+     *
+     * @return a context object or null if there isn't one
+     */
     @SuppressWarnings("unchecked")
     public <T> T getContext() {
         return (T) context;
     }
 
+    /**
+     * Each call to {@link org.dataloader.DataLoader#load(Object, Object)} or
+     * {@link org.dataloader.DataLoader#loadMany(java.util.List, java.util.List)} can be given
+     * a context object when it is invoked.  A map of them is present by this method.
+     *
+     * @return a map of key context objects
+     */
+    public Map<Object, Object> getKeyContexts() {
+        return keyContexts;
+    }
+
     public static Builder newBatchLoaderEnvironment() {
         return new Builder();
     }
 
     public static class Builder {
         private Object context;
+        private Map<Object, Object> keyContexts = Collections.emptyMap();
 
         private Builder() {
 
@@ -33,8 +59,13 @@ public Builder context(Object context) {
             return this;
         }
 
+        public Builder keyContexts(Map<Object, Object> keyContexts) {
+            this.keyContexts = nonNull(keyContexts);
+            return this;
+        }
+
         public BatchLoaderEnvironment build() {
-            return new BatchLoaderEnvironment(context);
+            return new BatchLoaderEnvironment(context, keyContexts);
         }
     }
 }
diff --git a/src/main/java/org/dataloader/DataLoader.java b/src/main/java/org/dataloader/DataLoader.java
index fa86d2b..14508e2 100644
--- a/src/main/java/org/dataloader/DataLoader.java
+++ b/src/main/java/org/dataloader/DataLoader.java
@@ -20,11 +20,10 @@
 import org.dataloader.stats.Statistics;
 import org.dataloader.stats.StatisticsCollector;
 
-import java.util.AbstractMap.SimpleImmutableEntry;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
-import java.util.stream.Collectors;
 
 import static org.dataloader.impl.Assertions.nonNull;
 
@@ -60,7 +59,6 @@ public class DataLoader<K, V> {
     private final DataLoaderHelper<K, V> helper;
     private final DataLoaderOptions loaderOptions;
     private final CacheMap<Object, CompletableFuture<V>> futureCache;
-    private final List<SimpleImmutableEntry<K, CompletableFuture<V>>> loaderQueue;
     private final StatisticsCollector stats;
 
     /**
@@ -237,6 +235,7 @@ public static <K, V> DataLoader<K, V> newMappedDataLoader(MappedBatchLoader<K, V
      * Using Try objects allows you to capture a value returned or an exception that might
      * have occurred trying to get a value. .
      * <p>
+     *
      * @param batchLoadFunction the batch load function to use that uses {@link org.dataloader.Try} objects
      * @param <K>               the key type
      * @param <V>               the value type
@@ -357,10 +356,9 @@ private DataLoader(Object batchLoadFunction, DataLoaderOptions options) {
         this.loaderOptions = options == null ? new DataLoaderOptions() : options;
         this.futureCache = determineCacheMap(loaderOptions);
         // order of keys matter in data loader
-        this.loaderQueue = new ArrayList<>();
         this.stats = nonNull(this.loaderOptions.getStatisticsCollector());
 
-        this.helper = new DataLoaderHelper<>(this, batchLoadFunction, this.loaderOptions, this.futureCache, this.loaderQueue, this.stats);
+        this.helper = new DataLoaderHelper<>(this, batchLoadFunction, this.loaderOptions, this.futureCache, this.stats);
     }
 
     @SuppressWarnings("unchecked")
@@ -380,7 +378,26 @@ private CacheMap<Object, CompletableFuture<V>> determineCacheMap(DataLoaderOptio
      * @return the future of the value
      */
     public CompletableFuture<V> load(K key) {
-        return helper.load(key);
+        return load(key, null);
+    }
+
+    /**
+     * Requests to load the data with the specified key asynchronously, and returns a future of the resulting value.
+     * <p>
+     * If batching is enabled (the default), you'll have to call {@link DataLoader#dispatch()} at a later stage to
+     * start batch execution. If you forget this call the future will never be completed (unless already completed,
+     * and returned from cache).
+     * <p>
+     * The key context object may be useful in the batch loader interfaces such as {@link org.dataloader.BatchLoaderWithContext} or
+     * {@link org.dataloader.MappedBatchLoaderWithContext} to help retrieve data.
+     *
+     * @param key        the key to load
+     * @param keyContext a context object that is specific to this key
+     *
+     * @return the future of the value
+     */
+    public CompletableFuture<V> load(K key, Object keyContext) {
+        return helper.load(key, keyContext);
     }
 
     /**
@@ -396,11 +413,39 @@ public CompletableFuture<V> load(K key) {
      * @return the composite future of the list of values
      */
     public CompletableFuture<List<V>> loadMany(List<K> keys) {
-        synchronized (this) {
-            List<CompletableFuture<V>> collect = keys.stream()
-                    .map(this::load)
-                    .collect(Collectors.toList());
+        return loadMany(keys, Collections.emptyList());
+    }
 
+    /**
+     * Requests to load the list of data provided by the specified keys asynchronously, and returns a composite future
+     * of the resulting values.
+     * <p>
+     * If batching is enabled (the default), you'll have to call {@link DataLoader#dispatch()} at a later stage to
+     * start batch execution. If you forget this call the future will never be completed (unless already completed,
+     * and returned from cache).
+     * <p>
+     * The key context object may be useful in the batch loader interfaces such as {@link org.dataloader.BatchLoaderWithContext} or
+     * {@link org.dataloader.MappedBatchLoaderWithContext} to help retrieve data.
+     *
+     * @param keys        the list of keys to load
+     * @param keyContexts the list of key calling context objects
+     *
+     * @return the composite future of the list of values
+     */
+    public CompletableFuture<List<V>> loadMany(List<K> keys, List<Object> keyContexts) {
+        nonNull(keys);
+        nonNull(keyContexts);
+
+        synchronized (this) {
+            List<CompletableFuture<V>> collect = new ArrayList<>();
+            for (int i = 0; i < keys.size(); i++) {
+                K key = keys.get(i);
+                Object keyContext = null;
+                if (i < keyContexts.size()) {
+                    keyContext = keyContexts.get(i);
+                }
+                collect.add(load(key, keyContext));
+            }
             return CompletableFutureKit.allOf(collect);
         }
     }
@@ -441,9 +486,7 @@ public List<V> dispatchAndJoin() {
      * @return the depth of the batched key loads that need to be dispatched
      */
     public int dispatchDepth() {
-        synchronized (this) {
-            return loaderQueue.size();
-        }
+        return helper.dispatchDepth();
     }
 
 
diff --git a/src/main/java/org/dataloader/DataLoaderHelper.java b/src/main/java/org/dataloader/DataLoaderHelper.java
index 87e1ebc..56a97c7 100644
--- a/src/main/java/org/dataloader/DataLoaderHelper.java
+++ b/src/main/java/org/dataloader/DataLoaderHelper.java
@@ -3,9 +3,9 @@
 import org.dataloader.impl.CompletableFutureKit;
 import org.dataloader.stats.StatisticsCollector;
 
-import java.util.AbstractMap;
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
@@ -26,23 +26,48 @@
  */
 class DataLoaderHelper<K, V> {
 
-    private final DataLoader<K,V> dataLoader;
+    class LoaderQueueEntry<K, V> {
+
+        final K key;
+        final V value;
+        final Object callContext;
+
+        public LoaderQueueEntry(K key, V value, Object callContext) {
+            this.key = key;
+            this.value = value;
+            this.callContext = callContext;
+        }
+
+        K getKey() {
+            return key;
+        }
+
+        V getValue() {
+            return value;
+        }
+
+        Object getCallContext() {
+            return callContext;
+        }
+    }
+
+    private final DataLoader<K, V> dataLoader;
     private final Object batchLoadFunction;
     private final DataLoaderOptions loaderOptions;
     private final CacheMap<Object, CompletableFuture<V>> futureCache;
-    private final List<AbstractMap.SimpleImmutableEntry<K, CompletableFuture<V>>> loaderQueue;
+    private final List<LoaderQueueEntry<K, CompletableFuture<V>>> loaderQueue;
     private final StatisticsCollector stats;
 
-    DataLoaderHelper(DataLoader<K,V> dataLoader, Object batchLoadFunction, DataLoaderOptions loaderOptions, CacheMap<Object, CompletableFuture<V>> futureCache, List<AbstractMap.SimpleImmutableEntry<K, CompletableFuture<V>>> loaderQueue, StatisticsCollector stats) {
+    DataLoaderHelper(DataLoader<K, V> dataLoader, Object batchLoadFunction, DataLoaderOptions loaderOptions, CacheMap<Object, CompletableFuture<V>> futureCache, StatisticsCollector stats) {
         this.dataLoader = dataLoader;
         this.batchLoadFunction = batchLoadFunction;
         this.loaderOptions = loaderOptions;
         this.futureCache = futureCache;
-        this.loaderQueue = loaderQueue;
+        this.loaderQueue = new ArrayList<>();
         this.stats = stats;
     }
 
-    CompletableFuture<V> load(K key) {
+    CompletableFuture<V> load(K key, Object loadContext) {
         synchronized (dataLoader) {
             Object cacheKey = getCacheKey(nonNull(key));
             stats.incrementLoadCount();
@@ -59,11 +84,11 @@ CompletableFuture<V> load(K key) {
 
             CompletableFuture<V> future = new CompletableFuture<>();
             if (batchingEnabled) {
-                loaderQueue.add(new AbstractMap.SimpleImmutableEntry<>(key, future));
+                loaderQueue.add(new LoaderQueueEntry<>(key, future, loadContext));
             } else {
                 stats.incrementBatchLoadCountBy(1);
                 // immediate execution of batch function
-                future = invokeLoaderImmediately(key);
+                future = invokeLoaderImmediately(key, loadContext);
             }
             if (cachingEnabled) {
                 futureCache.set(cacheKey, future);
@@ -83,11 +108,13 @@ CompletableFuture<List<V>> dispatch() {
         //
         // we copy the pre-loaded set of futures ready for dispatch
         final List<K> keys = new ArrayList<>();
+        final List<Object> callContexts = new ArrayList<>();
         final List<CompletableFuture<V>> queuedFutures = new ArrayList<>();
         synchronized (dataLoader) {
             loaderQueue.forEach(entry -> {
                 keys.add(entry.getKey());
                 queuedFutures.add(entry.getValue());
+                callContexts.add(entry.getCallContext());
             });
             loaderQueue.clear();
         }
@@ -107,13 +134,13 @@ CompletableFuture<List<V>> dispatch() {
         //
         int maxBatchSize = loaderOptions.maxBatchSize();
         if (maxBatchSize > 0 && maxBatchSize < keys.size()) {
-            return sliceIntoBatchesOfBatches(keys, queuedFutures, maxBatchSize);
+            return sliceIntoBatchesOfBatches(keys, queuedFutures, callContexts, maxBatchSize);
         } else {
-            return dispatchQueueBatch(keys, queuedFutures);
+            return dispatchQueueBatch(keys, callContexts, queuedFutures);
         }
     }
 
-    private CompletableFuture<List<V>> sliceIntoBatchesOfBatches(List<K> keys, List<CompletableFuture<V>> queuedFutures, int maxBatchSize) {
+    private CompletableFuture<List<V>> sliceIntoBatchesOfBatches(List<K> keys, List<CompletableFuture<V>> queuedFutures, List<Object> callContexts, int maxBatchSize) {
         // the number of keys is > than what the batch loader function can accept
         // so make multiple calls to the loader
         List<CompletableFuture<List<V>>> allBatches = new ArrayList<>();
@@ -126,8 +153,9 @@ private CompletableFuture<List<V>> sliceIntoBatchesOfBatches(List<K> keys, List<
 
             List<K> subKeys = keys.subList(fromIndex, toIndex);
             List<CompletableFuture<V>> subFutures = queuedFutures.subList(fromIndex, toIndex);
+            List<Object> subCallContexts = callContexts.subList(fromIndex, toIndex);
 
-            allBatches.add(dispatchQueueBatch(subKeys, subFutures));
+            allBatches.add(dispatchQueueBatch(subKeys, subCallContexts, subFutures));
         }
         //
         // now reassemble all the futures into one that is the complete set of results
@@ -139,9 +167,9 @@ private CompletableFuture<List<V>> sliceIntoBatchesOfBatches(List<K> keys, List<
     }
 
     @SuppressWarnings("unchecked")
-    private CompletableFuture<List<V>> dispatchQueueBatch(List<K> keys, List<CompletableFuture<V>> queuedFutures) {
+    private CompletableFuture<List<V>> dispatchQueueBatch(List<K> keys, List<Object> callContexts, List<CompletableFuture<V>> queuedFutures) {
         stats.incrementBatchLoadCountBy(keys.size());
-        CompletionStage<List<V>> batchLoad = invokeLoader(keys);
+        CompletionStage<List<V>> batchLoad = invokeLoader(keys, callContexts);
         return batchLoad
                 .toCompletableFuture()
                 .thenApply(values -> {
@@ -190,11 +218,14 @@ private void assertResultSize(List<K> keys, List<V> values) {
     }
 
 
-    CompletableFuture<V> invokeLoaderImmediately(K key) {
+    CompletableFuture<V> invokeLoaderImmediately(K key, Object keyContext) {
         List<K> keys = singletonList(key);
         CompletionStage<V> singleLoadCall;
         try {
-            BatchLoaderEnvironment environment = loaderOptions.getBatchLoaderEnvironmentProvider().get();
+            Object context = loaderOptions.getBatchLoaderEnvironmentProvider().getContext();
+            Map<Object, Object> keyContextMap = mkKeyContextMap(keys, singletonList(keyContext));
+            BatchLoaderEnvironment environment = BatchLoaderEnvironment.newBatchLoaderEnvironment()
+                    .context(context).keyContexts(keyContextMap).build();
             if (isMapLoader()) {
                 singleLoadCall = invokeMapBatchLoader(keys, environment).thenApply(list -> list.get(0));
             } else {
@@ -206,10 +237,13 @@ CompletableFuture<V> invokeLoaderImmediately(K key) {
         }
     }
 
-    CompletionStage<List<V>> invokeLoader(List<K> keys) {
+    CompletionStage<List<V>> invokeLoader(List<K> keys, List<Object> keyContexts) {
         CompletionStage<List<V>> batchLoad;
         try {
-            BatchLoaderEnvironment environment = loaderOptions.getBatchLoaderEnvironmentProvider().get();
+            Object context = loaderOptions.getBatchLoaderEnvironmentProvider().getContext();
+            Map<Object, Object> keyContextMap = mkKeyContextMap(keys, keyContexts);
+            BatchLoaderEnvironment environment = BatchLoaderEnvironment.newBatchLoaderEnvironment()
+                    .context(context).keyContexts(keyContextMap).build();
             if (isMapLoader()) {
                 batchLoad = invokeMapBatchLoader(keys, environment);
             } else {
@@ -232,11 +266,11 @@ private CompletionStage<List<V>> invokeListBatchLoader(List<K> keys, BatchLoader
         return nonNull(loadResult, "Your batch loader function MUST return a non null CompletionStage promise");
     }
 
+
     /*
      * Turns a map of results that MAY be smaller than the key list back into a list by mapping null
      * to missing elements.
      */
-
     @SuppressWarnings("unchecked")
     private CompletionStage<List<V>> invokeMapBatchLoader(List<K> keys, BatchLoaderEnvironment environment) {
         CompletionStage<Map<K, V>> loadResult;
@@ -259,4 +293,25 @@ private CompletionStage<List<V>> invokeMapBatchLoader(List<K> keys, BatchLoaderE
     private boolean isMapLoader() {
         return batchLoadFunction instanceof MappedBatchLoader || batchLoadFunction instanceof MappedBatchLoaderWithContext;
     }
+
+    private Map<Object, Object> mkKeyContextMap(List<K> keys, List<Object> keyContexts) {
+        Map<Object, Object> map = new HashMap<>();
+        for (int i = 0; i < keys.size(); i++) {
+            K key = keys.get(i);
+            Object keyContext = null;
+            if (i < keyContexts.size()) {
+                keyContext = keyContexts.get(i);
+            }
+            if (keyContext != null) {
+                map.put(key, keyContext);
+            }
+        }
+        return map;
+    }
+
+    int dispatchDepth() {
+        synchronized (dataLoader) {
+            return loaderQueue.size();
+        }
+    }
 }
diff --git a/src/main/java/org/dataloader/DataLoaderOptions.java b/src/main/java/org/dataloader/DataLoaderOptions.java
index 8cab094..c8710b7 100644
--- a/src/main/java/org/dataloader/DataLoaderOptions.java
+++ b/src/main/java/org/dataloader/DataLoaderOptions.java
@@ -31,7 +31,7 @@
  */
 public class DataLoaderOptions {
 
-    private static final BatchLoaderEnvironmentProvider NULL_PROVIDER = () -> BatchLoaderEnvironment.newBatchLoaderEnvironment().build();
+    private static final BatchLoaderContextProvider NULL_PROVIDER = () -> null;
 
     private boolean batchingEnabled;
     private boolean cachingEnabled;
@@ -39,7 +39,7 @@ public class DataLoaderOptions {
     private CacheMap cacheMap;
     private int maxBatchSize;
     private Supplier<StatisticsCollector> statisticsCollector;
-    private BatchLoaderEnvironmentProvider environmentProvider;
+    private BatchLoaderContextProvider environmentProvider;
 
     /**
      * Creates a new data loader options with default settings.
@@ -210,7 +210,7 @@ public DataLoaderOptions setStatisticsCollector(Supplier<StatisticsCollector> st
     /**
      * @return the batch environment provider that will be used to give context to batch load functions
      */
-    public BatchLoaderEnvironmentProvider getBatchLoaderEnvironmentProvider() {
+    public BatchLoaderContextProvider getBatchLoaderEnvironmentProvider() {
         return environmentProvider;
     }
 
@@ -221,7 +221,7 @@ public BatchLoaderEnvironmentProvider getBatchLoaderEnvironmentProvider() {
      *
      * @return the data loader options for fluent coding
      */
-    public DataLoaderOptions setBatchLoaderEnvironmentProvider(BatchLoaderEnvironmentProvider environmentProvider) {
+    public DataLoaderOptions setBatchLoaderEnvironmentProvider(BatchLoaderContextProvider environmentProvider) {
         this.environmentProvider = nonNull(environmentProvider);
         return this;
     }
diff --git a/src/test/java/ReadmeExamples.java b/src/test/java/ReadmeExamples.java
index 878943b..a3345e6 100644
--- a/src/test/java/ReadmeExamples.java
+++ b/src/test/java/ReadmeExamples.java
@@ -84,11 +84,8 @@ public CompletionStage<List<User>> load(List<Long> userIds) {
     }
 
     private void callContextExample() {
-        BatchLoaderEnvironment batchLoaderEnvironment = BatchLoaderEnvironment.newBatchLoaderEnvironment()
-                .context(SecurityCtx.getCallingUserCtx()).build();
-
         DataLoaderOptions options = DataLoaderOptions.newOptions()
-                .setBatchLoaderEnvironmentProvider(() -> batchLoaderEnvironment);
+                .setBatchLoaderEnvironmentProvider(() -> SecurityCtx.getCallingUserCtx());
 
         BatchLoaderWithContext<String, String> batchLoader = new BatchLoaderWithContext<String, String>() {
             @Override
diff --git a/src/test/java/org/dataloader/DataLoaderBatchLoaderEnvironmentTest.java b/src/test/java/org/dataloader/DataLoaderBatchLoaderEnvironmentTest.java
index 88c2a4e..ac171bc 100644
--- a/src/test/java/org/dataloader/DataLoaderBatchLoaderEnvironmentTest.java
+++ b/src/test/java/org/dataloader/DataLoaderBatchLoaderEnvironmentTest.java
@@ -2,6 +2,8 @@
 
 import org.junit.Test;
 
+import java.util.ArrayList;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -9,7 +11,7 @@
 import java.util.stream.Collectors;
 
 import static java.util.Arrays.asList;
-import static org.dataloader.BatchLoaderEnvironment.newBatchLoaderEnvironment;
+import static java.util.Collections.singletonList;
 import static org.hamcrest.Matchers.equalTo;
 import static org.junit.Assert.assertThat;
 
@@ -18,6 +20,18 @@
  */
 public class DataLoaderBatchLoaderEnvironmentTest {
 
+    private BatchLoaderWithContext<String, String> contextBatchLoader() {
+        return (keys, environment) -> {
+            List<String> list = keys.stream().map(k -> {
+                Object context = environment.getContext();
+                Object keyContext = environment.getKeyContexts().get(k);
+                return k + "-" + context + "-" + keyContext;
+            }).collect(Collectors.toList());
+            return CompletableFuture.completedFuture(list);
+        };
+    }
+
+
     @Test
     public void context_is_passed_to_batch_loader_function() throws Exception {
         BatchLoaderWithContext<String, String> batchLoader = (keys, environment) -> {
@@ -25,7 +39,7 @@ public void context_is_passed_to_batch_loader_function() throws Exception {
             return CompletableFuture.completedFuture(list);
         };
         DataLoaderOptions options = DataLoaderOptions.newOptions()
-                .setBatchLoaderEnvironmentProvider(() -> newBatchLoaderEnvironment().context("ctx").build());
+                .setBatchLoaderEnvironmentProvider(() -> "ctx");
         DataLoader<String, String> loader = DataLoader.newDataLoader(batchLoader, options);
 
         loader.load("A");
@@ -37,24 +51,78 @@ public void context_is_passed_to_batch_loader_function() throws Exception {
         assertThat(results, equalTo(asList("A-ctx", "B-ctx", "C-ctx", "D-ctx")));
     }
 
+    @Test
+    public void key_contexts_are_passed_to_batch_loader_function() throws Exception {
+        BatchLoaderWithContext<String, String> batchLoader = contextBatchLoader();
+        DataLoaderOptions options = DataLoaderOptions.newOptions()
+                .setBatchLoaderEnvironmentProvider(() -> "ctx");
+        DataLoader<String, String> loader = DataLoader.newDataLoader(batchLoader, options);
+
+        loader.load("A", "aCtx");
+        loader.load("B", "bCtx");
+        loader.loadMany(asList("C", "D"), asList("cCtx", "dCtx"));
+
+        List<String> results = loader.dispatchAndJoin();
+
+        assertThat(results, equalTo(asList("A-ctx-aCtx", "B-ctx-bCtx", "C-ctx-cCtx", "D-ctx-dCtx")));
+    }
+
+    @Test
+    public void key_contexts_are_passed_to_batch_loader_function_when_batching_disabled() throws Exception {
+        BatchLoaderWithContext<String, String> batchLoader = contextBatchLoader();
+        DataLoaderOptions options = DataLoaderOptions.newOptions()
+                .setBatchingEnabled(false)
+                .setBatchLoaderEnvironmentProvider(() -> "ctx");
+        DataLoader<String, String> loader = DataLoader.newDataLoader(batchLoader, options);
+
+        CompletableFuture<String> aLoad = loader.load("A", "aCtx");
+        CompletableFuture<String> bLoad = loader.load("B", "bCtx");
+        CompletableFuture<List<String>> canDLoad = loader.loadMany(asList("C", "D"), asList("cCtx", "dCtx"));
+
+        List<String> results = new ArrayList<>(asList(aLoad.join(), bLoad.join()));
+        results.addAll(canDLoad.join());
+
+        assertThat(results, equalTo(asList("A-ctx-aCtx", "B-ctx-bCtx", "C-ctx-cCtx", "D-ctx-dCtx")));
+    }
+
+    @Test
+    public void missing_key_contexts_are_passed_to_batch_loader_function() throws Exception {
+        BatchLoaderWithContext<String, String> batchLoader = contextBatchLoader();
+        DataLoaderOptions options = DataLoaderOptions.newOptions()
+                .setBatchLoaderEnvironmentProvider(() -> "ctx");
+        DataLoader<String, String> loader = DataLoader.newDataLoader(batchLoader, options);
+
+        loader.load("A", "aCtx");
+        loader.load("B");
+        loader.loadMany(asList("C", "D"), singletonList("cCtx"));
+
+        List<String> results = loader.dispatchAndJoin();
+
+        assertThat(results, equalTo(asList("A-ctx-aCtx", "B-ctx-null", "C-ctx-cCtx", "D-ctx-null")));
+    }
+
     @Test
     public void context_is_passed_to_map_batch_loader_function() throws Exception {
         MappedBatchLoaderWithContext<String, String> mapBatchLoader = (keys, environment) -> {
             Map<String, String> map = new HashMap<>();
-            keys.forEach(k -> map.put(k, k + "-" + environment.getContext()));
+            keys.forEach(k -> {
+                Object context = environment.getContext();
+                Object keyContext = environment.getKeyContexts().get(k);
+                map.put(k, k + "-" + context + "-" + keyContext);
+            });
             return CompletableFuture.completedFuture(map);
         };
         DataLoaderOptions options = DataLoaderOptions.newOptions()
-                .setBatchLoaderEnvironmentProvider(() -> newBatchLoaderEnvironment().context("ctx").build());
+                .setBatchLoaderEnvironmentProvider(() -> "ctx");
         DataLoader<String, String> loader = DataLoader.newMappedDataLoader(mapBatchLoader, options);
 
-        loader.load("A");
+        loader.load("A", "aCtx");
         loader.load("B");
-        loader.loadMany(asList("C", "D"));
+        loader.loadMany(asList("C", "D"), singletonList("cCtx"));
 
         List<String> results = loader.dispatchAndJoin();
 
-        assertThat(results, equalTo(asList("A-ctx", "B-ctx", "C-ctx", "D-ctx")));
+        assertThat(results, equalTo(asList("A-ctx-aCtx", "B-ctx-null", "C-ctx-cCtx", "D-ctx-null")));
     }
 
     @Test