diff --git a/gcloud-java-storage/src/main/java/com/google/cloud/storage/BatchRequest.java b/gcloud-java-storage/src/main/java/com/google/cloud/storage/BatchRequest.java
deleted file mode 100644
index 05dfe8e2a1dc..000000000000
--- a/gcloud-java-storage/src/main/java/com/google/cloud/storage/BatchRequest.java
+++ /dev/null
@@ -1,131 +0,0 @@
-/*
- * Copyright 2015 Google Inc. All Rights Reserved.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *       http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.google.cloud.storage;
-
-import com.google.cloud.storage.Storage.BlobGetOption;
-import com.google.cloud.storage.Storage.BlobSourceOption;
-import com.google.cloud.storage.Storage.BlobTargetOption;
-import com.google.common.collect.ImmutableMap;
-import com.google.common.collect.Lists;
-
-import java.io.Serializable;
-import java.util.LinkedHashMap;
-import java.util.Map;
-import java.util.Objects;
-
-/**
- * Google storage batch request.
- */
-public final class BatchRequest implements Serializable {
-
-  private static final long serialVersionUID = -1527992265939800345L;
-
-  private final Map<BlobId, Iterable<BlobSourceOption>> toDelete;
-  private final Map<BlobInfo, Iterable<BlobTargetOption>> toUpdate;
-  private final Map<BlobId, Iterable<BlobGetOption>> toGet;
-
-  public static class Builder {
-
-    private Map<BlobId, Iterable<BlobSourceOption>> toDelete = new LinkedHashMap<>();
-    private Map<BlobInfo, Iterable<BlobTargetOption>> toUpdate = new LinkedHashMap<>();
-    private Map<BlobId, Iterable<BlobGetOption>> toGet = new LinkedHashMap<>();
-
-    private Builder() {}
-
-    /**
-     * Delete the given blob.
-     */
-    public Builder delete(String bucket, String blob, BlobSourceOption... options) {
-      toDelete.put(BlobId.of(bucket, blob), Lists.newArrayList(options));
-      return this;
-    }
-
-    /**
-     * Delete the given blob.
-     */
-    public Builder delete(BlobId blob, BlobSourceOption... options) {
-      toDelete.put(blob, Lists.newArrayList(options));
-      return this;
-    }
-
-    /**
-     * Update the given blob.
-     */
-    public Builder update(BlobInfo blobInfo, BlobTargetOption... options) {
-      toUpdate.put(blobInfo, Lists.newArrayList(options));
-      return this;
-    }
-
-    /**
-     * Retrieve metadata for the given blob.
-     */
-    public Builder get(String bucket, String blob, BlobGetOption... options) {
-      toGet.put(BlobId.of(bucket, blob), Lists.newArrayList(options));
-      return this;
-    }
-
-    /**
-     * Retrieve metadata for the given blob.
-     */
-    public Builder get(BlobId blob, BlobGetOption... options) {
-      toGet.put(blob, Lists.newArrayList(options));
-      return this;
-    }
-
-    public BatchRequest build() {
-      return new BatchRequest(this);
-    }
-  }
-
-  private BatchRequest(Builder builder) {
-    toDelete = ImmutableMap.copyOf(builder.toDelete);
-    toUpdate = ImmutableMap.copyOf(builder.toUpdate);
-    toGet = ImmutableMap.copyOf(builder.toGet);
-  }
-
-  @Override
-  public int hashCode() {
-    return Objects.hash(toDelete, toUpdate, toGet);
-  }
-
-  @Override
-  public boolean equals(Object obj) {
-    if (!(obj instanceof BatchRequest)) {
-      return false;
-    }
-    BatchRequest other = (BatchRequest) obj;
-    return Objects.equals(toDelete, other.toDelete)
-        && Objects.equals(toUpdate, other.toUpdate)
-        && Objects.equals(toGet, other.toGet);
-  }
-
-  public Map<BlobId, Iterable<BlobSourceOption>> toDelete() {
-    return toDelete;
-  }
-
-  public Map<BlobInfo, Iterable<BlobTargetOption>> toUpdate() {
-    return toUpdate;
-  }
-
-  public Map<BlobId, Iterable<BlobGetOption>> toGet() {
-    return toGet;
-  }
-
-  public static Builder builder() {
-    return new Builder();
-  }
-}
diff --git a/gcloud-java-storage/src/main/java/com/google/cloud/storage/BatchResponse.java b/gcloud-java-storage/src/main/java/com/google/cloud/storage/BatchResponse.java
deleted file mode 100644
index d1e56758b9d2..000000000000
--- a/gcloud-java-storage/src/main/java/com/google/cloud/storage/BatchResponse.java
+++ /dev/null
@@ -1,162 +0,0 @@
-/*
- * Copyright 2015 Google Inc. All Rights Reserved.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *       http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.google.cloud.storage;
-
-import com.google.common.base.MoreObjects;
-import com.google.common.collect.ImmutableList;
-
-import java.io.Serializable;
-import java.util.List;
-import java.util.Objects;
-
-/**
- * Google Storage batch response.
- */
-public class BatchResponse implements Serializable {
-
-  private static final long serialVersionUID = 1057416839397037706L;
-
-  private final List<Result<Boolean>> deleteResult;
-  private final List<Result<Blob>> updateResult;
-  private final List<Result<Blob>> getResult;
-
-  public static class Result<T extends Serializable> implements Serializable {
-
-    private static final long serialVersionUID = -1946539570170529094L;
-    private static final Result<Serializable> EMPTY = Result.of(null);
-
-    private final T value;
-    private final StorageException exception;
-
-
-    public Result(T value) {
-      this.value = value;
-      this.exception = null;
-    }
-
-    public Result(StorageException exception) {
-      this.exception = exception;
-      this.value = null;
-    }
-
-    static <T extends Serializable> Result<T> of(T value) {
-      return new Result<>(value);
-    }
-
-    /**
-     * Returns the result.
-     *
-     * @throws StorageException if failed
-     */
-    public T get() throws StorageException {
-      if (failed()) {
-        throw failure();
-      }
-      return value;
-    }
-
-    /**
-     * Returns the failure or {@code null} if was successful.
-     */
-    public StorageException failure() {
-      return exception;
-    }
-
-    /**
-     * Returns {@code true} if failed, {@code false} otherwise.
-     */
-    public boolean failed() {
-      return exception != null;
-    }
-
-    @Override
-    public int hashCode() {
-      return Objects.hash(value, exception);
-    }
-
-    @Override
-    public boolean equals(Object obj) {
-      if (!(obj instanceof Result)) {
-        return false;
-      }
-      Result<?> other = (Result<?>) obj;
-      return Objects.equals(value, other.value)
-          && Objects.equals(exception, other.exception);
-    }
-
-    @Override
-    public String toString() {
-      return MoreObjects.toStringHelper(this)
-          .add("value", value)
-          .add("exception", exception)
-          .toString();
-    }
-
-    static <T extends Serializable> Result<T> empty() {
-      @SuppressWarnings("unchecked")
-      Result<T> result = (Result<T>) EMPTY;
-      return result;
-    }
-  }
-
-  BatchResponse(List<Result<Boolean>> deleteResult, List<Result<Blob>> updateResult,
-      List<Result<Blob>> getResult) {
-    this.deleteResult = ImmutableList.copyOf(deleteResult);
-    this.updateResult = ImmutableList.copyOf(updateResult);
-    this.getResult = ImmutableList.copyOf(getResult);
-  }
-
-  @Override
-  public final int hashCode() {
-    return Objects.hash(deleteResult, updateResult, getResult);
-  }
-
-  @Override
-  public final boolean equals(Object obj) {
-    if (obj == this) {
-      return true;
-    }
-    if (obj == null || !obj.getClass().equals(BatchResponse.class)) {
-      return false;
-    }
-    BatchResponse other = (BatchResponse) obj;
-    return Objects.equals(deleteResult, other.deleteResult)
-        && Objects.equals(updateResult, other.updateResult)
-        && Objects.equals(getResult, other.getResult);
-  }
-
-  /**
-   * Returns the results for the delete operations using the request order.
-   */
-  public List<Result<Boolean>> deletes() {
-    return deleteResult;
-  }
-
-  /**
-   * Returns the results for the update operations using the request order.
-   */
-  public List<Result<Blob>> updates() {
-    return updateResult;
-  }
-
-  /**
-   * Returns the results for the get operations using the request order.
-   */
-  public List<Result<Blob>> gets() {
-    return getResult;
-  }
-}
diff --git a/gcloud-java-storage/src/main/java/com/google/cloud/storage/Bucket.java b/gcloud-java-storage/src/main/java/com/google/cloud/storage/Bucket.java
index 9f5a2e2499a0..ebb6835eba4b 100644
--- a/gcloud-java-storage/src/main/java/com/google/cloud/storage/Bucket.java
+++ b/gcloud-java-storage/src/main/java/com/google/cloud/storage/Bucket.java
@@ -26,6 +26,7 @@
 import com.google.cloud.storage.Storage.BucketTargetOption;
 import com.google.cloud.storage.spi.StorageRpc;
 import com.google.common.base.Function;
+import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Sets;
 
@@ -33,9 +34,7 @@
 import java.io.InputStream;
 import java.io.ObjectInputStream;
 import java.io.Serializable;
-import java.util.ArrayList;
 import java.util.Arrays;
-import java.util.Collections;
 import java.util.List;
 import java.util.Objects;
 import java.util.Set;
@@ -609,19 +608,28 @@ public Blob get(String blob, BlobGetOption... options) {
    * @throws StorageException upon failure
    */
   public List<Blob> get(String blobName1, String blobName2, String... blobNames) {
-    BatchRequest.Builder batch = BatchRequest.builder();
-    batch.get(name(), blobName1);
-    batch.get(name(), blobName2);
-    for (String name : blobNames) {
-      batch.get(name(), name);
-    }
-    List<Blob> blobs = new ArrayList<>(blobNames.length);
-    BatchResponse response = storage.submit(batch.build());
-    for (BatchResponse.Result<Blob> result : response.gets()) {
-      BlobInfo blobInfo = result.get();
-      blobs.add(blobInfo != null ? new Blob(storage, new BlobInfo.BuilderImpl(blobInfo)) : null);
-    }
-    return Collections.unmodifiableList(blobs);
+    List<BlobId> blobIds = Lists.newArrayListWithCapacity(blobNames.length + 2);
+    blobIds.add(BlobId.of(name(), blobName1));
+    blobIds.add(BlobId.of(name(), blobName2));
+    for (String blobName : blobNames) {
+      blobIds.add(BlobId.of(name(), blobName));
+    }
+    return storage.get(blobIds);
+  }
+
+  /**
+   * Returns a list of requested blobs in this bucket. Blobs that do not exist are null.
+   *
+   * @param blobNames blobs to get
+   * @return an immutable list of {@code Blob} objects
+   * @throws StorageException upon failure
+   */
+  public List<Blob> get(Iterable<String> blobNames) {
+    ImmutableList.Builder<BlobId> builder = ImmutableList.builder();
+    for (String blobName : blobNames) {
+      builder.add(BlobId.of(name(), blobName));
+    }
+    return storage.get(builder.build());
   }
 
   /**
diff --git a/gcloud-java-storage/src/main/java/com/google/cloud/storage/Storage.java b/gcloud-java-storage/src/main/java/com/google/cloud/storage/Storage.java
index 4c32aea1a428..b1087f72bce1 100644
--- a/gcloud-java-storage/src/main/java/com/google/cloud/storage/Storage.java
+++ b/gcloud-java-storage/src/main/java/com/google/cloud/storage/Storage.java
@@ -1425,12 +1425,29 @@ public static Builder builder() {
   byte[] readAllBytes(BlobId blob, BlobSourceOption... options);
 
   /**
-   * Sends a batch request.
+   * Creates a new empty batch for grouping multiple service calls in one underlying RPC call.
    *
-   * @return the batch response
-   * @throws StorageException upon failure
+   * <p>Example of using a batch request to delete, update and get a blob:
+   * <pre>{@code
+   * StorageBatch batch = storage.batch();
+   * BlobId firstBlob = BlobId.of("bucket", "blob1"));
+   * BlobId secondBlob = BlobId.of("bucket", "blob2"));
+   * batch.delete(firstBlob).notify(new BatchResult.Callback<Boolean, StorageException>() {
+   *   public void success(Boolean result) {
+   *     // deleted successfully
+   *   }
+   *
+   *   public void error(StorageException exception) {
+   *     // delete failed
+   *   }
+   * });
+   * batch.update(BlobInfo.builder(secondBlob).contentType("text/plain").build());
+   * StorageBatchResult<Blob> result = batch.get(secondBlob);
+   * batch.submit();
+   * Blob blob = result.get(); // returns get result or throws StorageException
+   * }</pre>
    */
-  BatchResponse submit(BatchRequest batchRequest);
+  StorageBatch batch();
 
   /**
    * Returns a channel for reading the blob's content. The blob's latest generation is read. If the
@@ -1534,6 +1551,16 @@ public static Builder builder() {
    */
   List<Blob> get(BlobId... blobIds);
 
+  /**
+   * Gets the requested blobs. A batch request is used to perform this call.
+   *
+   * @param blobIds blobs to get
+   * @return an immutable list of {@code Blob} objects. If a blob does not exist or access to it
+   *     has been denied the corresponding item in the list is {@code null}.
+   * @throws StorageException upon failure
+   */
+  List<Blob> get(Iterable<BlobId> blobIds);
+
   /**
    * Updates the requested blobs. A batch request is used to perform this call. Original metadata
    * are merged with metadata in the provided {@code BlobInfo} objects. To replace metadata instead
@@ -1548,6 +1575,20 @@ public static Builder builder() {
    */
   List<Blob> update(BlobInfo... blobInfos);
 
+  /**
+   * Updates the requested blobs. A batch request is used to perform this call. Original metadata
+   * are merged with metadata in the provided {@code BlobInfo} objects. To replace metadata instead
+   * you first have to unset them. Unsetting metadata can be done by setting the provided
+   * {@code BlobInfo} objects metadata to {@code null}. See
+   * {@link #update(BlobInfo)} for a code example.
+   *
+   * @param blobInfos blobs to update
+   * @return an immutable list of {@code Blob} objects. If a blob does not exist or access to it
+   *     has been denied the corresponding item in the list is {@code null}.
+   * @throws StorageException upon failure
+   */
+  List<Blob> update(Iterable<BlobInfo> blobInfos);
+
   /**
    * Deletes the requested blobs. A batch request is used to perform this call.
    *
@@ -1558,4 +1599,15 @@ public static Builder builder() {
    * @throws StorageException upon failure
    */
   List<Boolean> delete(BlobId... blobIds);
+
+  /**
+   * Deletes the requested blobs. A batch request is used to perform this call.
+   *
+   * @param blobIds blobs to delete
+   * @return an immutable list of booleans. If a blob has been deleted the corresponding item in the
+   *     list is {@code true}. If a blob was not found, deletion failed or access to the resource
+   *     was denied the corresponding item is {@code false}.
+   * @throws StorageException upon failure
+   */
+  List<Boolean> delete(Iterable<BlobId> blobIds);
 }
diff --git a/gcloud-java-storage/src/main/java/com/google/cloud/storage/StorageBatch.java b/gcloud-java-storage/src/main/java/com/google/cloud/storage/StorageBatch.java
new file mode 100644
index 000000000000..1153d8015453
--- /dev/null
+++ b/gcloud-java-storage/src/main/java/com/google/cloud/storage/StorageBatch.java
@@ -0,0 +1,207 @@
+/*
+ * Copyright 2016 Google Inc. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.storage;
+
+import static java.net.HttpURLConnection.HTTP_NOT_FOUND;
+
+import com.google.api.client.googleapis.json.GoogleJsonError;
+import com.google.api.services.storage.model.StorageObject;
+import com.google.cloud.storage.Storage.BlobGetOption;
+import com.google.cloud.storage.Storage.BlobSourceOption;
+import com.google.cloud.storage.Storage.BlobTargetOption;
+import com.google.cloud.storage.spi.RpcBatch;
+import com.google.cloud.storage.spi.StorageRpc;
+import com.google.common.annotations.VisibleForTesting;
+
+import java.util.Map;
+
+/**
+ * A batch of operations to be submitted to Google Cloud Storage using a single RPC request.
+ *
+ * <p>Example of using a batch request to delete, update and get a blob:
+ * <pre>{@code
+ * StorageBatch batch = storage.batch();
+ * BlobId firstBlob = BlobId.of("bucket", "blob1"));
+ * BlobId secondBlob = BlobId.of("bucket", "blob2"));
+ * batch.delete(firstBlob).notify(new BatchResult.Callback<Boolean, StorageException>() {
+ *   public void success(Boolean result) {
+ *     // deleted successfully
+ *   }
+ *
+ *   public void error(StorageException exception) {
+ *     // delete failed
+ *   }
+ * });
+ * batch.update(BlobInfo.builder(secondBlob).contentType("text/plain").build());
+ * StorageBatchResult<Blob> result = batch.get(secondBlob);
+ * batch.submit();
+ * Blob blob = result.get(); // returns get result or throws StorageException
+ * }</pre>
+ */
+public class StorageBatch {
+
+  private final RpcBatch batch;
+  private final StorageRpc storageRpc;
+  private final StorageOptions options;
+
+  StorageBatch(StorageOptions options) {
+    this.options = options;
+    this.storageRpc = options.rpc();
+    this.batch = storageRpc.createBatch();
+  }
+
+  @VisibleForTesting
+  Object batch() {
+    return batch;
+  }
+
+  @VisibleForTesting
+  StorageRpc storageRpc() {
+    return storageRpc;
+  }
+
+  @VisibleForTesting
+  StorageOptions options() {
+    return options;
+  }
+
+  /**
+   * Adds a request representing the "delete blob" operation to this batch. Calling {@link
+   * StorageBatchResult#get()} on the return value yields {@code true} upon successful deletion,
+   * {@code false} if the blob was not found, or throws a {@link StorageException} if the operation
+   * failed.
+   */
+  public StorageBatchResult<Boolean> delete(String bucket, String blob,
+      BlobSourceOption... options) {
+    return delete(BlobId.of(bucket, blob), options);
+  }
+
+  /**
+   * Adds a request representing the "delete blob" operation to this batch. Calling {@link
+   * StorageBatchResult#get()} on the return value yields {@code true} upon successful deletion,
+   * {@code false} if the blob was not found, or throws a {@link StorageException} if the operation
+   * failed.
+   */
+  public StorageBatchResult<Boolean> delete(BlobId blob, BlobSourceOption... options) {
+    StorageBatchResult<Boolean> result = new StorageBatchResult<>();
+    RpcBatch.Callback<Void> callback = createDeleteCallback(result);
+    Map<StorageRpc.Option, ?> optionMap = StorageImpl.optionMap(blob, options);
+    batch.addDelete(blob.toPb(), callback, optionMap);
+    return result;
+  }
+
+  /**
+   * Adds a request representing the "update blob" operation to this batch. The {@code options} can
+   * be used in the same way as for {@link Storage#update(BlobInfo, BlobTargetOption...)}. Calling
+   * {@link StorageBatchResult#get()} on the return value yields the updated {@link Blob} if
+   * successful, or throws a {@link StorageException} if the operation failed.
+   */
+  public StorageBatchResult<Blob> update(BlobInfo blobInfo, BlobTargetOption... options) {
+    StorageBatchResult<Blob> result = new StorageBatchResult<>();
+    RpcBatch.Callback<StorageObject> callback = createUpdateCallback(this.options, result);
+    Map<StorageRpc.Option, ?> optionMap = StorageImpl.optionMap(blobInfo, options);
+    batch.addPatch(blobInfo.toPb(), callback, optionMap);
+    return result;
+  }
+
+  /**
+   * Adds a request representing the "get blob" operation to this batch. The {@code options} can be
+   * used in the same way as for {@link Storage#get(BlobId, BlobGetOption...)}. Calling
+   * {@link StorageBatchResult#get()} on the return value yields the requested {@link Blob} if
+   * successful, {@code null} if no such blob exists, or throws a {@link StorageException} if the
+   * operation failed.
+   */
+  public StorageBatchResult<Blob> get(String bucket, String blob, BlobGetOption... options) {
+    return get(BlobId.of(bucket, blob), options);
+  }
+
+  /**
+   * Adds a request representing the "get blob" operation to this batch. The {@code options} can be
+   * used in the same way as for {@link Storage#get(BlobId, BlobGetOption...)}. Calling
+   * {@link StorageBatchResult#get()} on the return value yields the requested {@link Blob} if
+   * successful, {@code null} if no such blob exists, or throws a {@link StorageException} if the
+   * operation failed.
+   */
+  public StorageBatchResult<Blob> get(BlobId blob, BlobGetOption... options) {
+    StorageBatchResult<Blob> result = new StorageBatchResult<>();
+    RpcBatch.Callback<StorageObject> callback = createGetCallback(this.options, result);
+    Map<StorageRpc.Option, ?> optionMap = StorageImpl.optionMap(blob, options);
+    batch.addGet(blob.toPb(), callback, optionMap);
+    return result;
+  }
+
+  /**
+   * Submits this batch for processing using a single RPC request.
+   */
+  public void submit() {
+    batch.submit();
+  }
+
+  private RpcBatch.Callback<Void> createDeleteCallback(final StorageBatchResult<Boolean> result) {
+    return new RpcBatch.Callback<Void>() {
+      @Override
+      public void onSuccess(Void response) {
+        result.success(true);
+      }
+
+      @Override
+      public void onFailure(GoogleJsonError googleJsonError) {
+        StorageException serviceException = new StorageException(googleJsonError);
+        if (serviceException.code() == HTTP_NOT_FOUND) {
+          result.success(false);
+        } else {
+          result.error(serviceException);
+        }
+      }
+    };
+  }
+
+  private RpcBatch.Callback<StorageObject> createGetCallback(final StorageOptions serviceOptions,
+      final StorageBatchResult<Blob> result) {
+    return new RpcBatch.Callback<StorageObject>() {
+      @Override
+      public void onSuccess(StorageObject response) {
+        result.success(response == null ? null : Blob.fromPb(serviceOptions.service(), response));
+      }
+
+      @Override
+      public void onFailure(GoogleJsonError googleJsonError) {
+        StorageException serviceException = new StorageException(googleJsonError);
+        if (serviceException.code() == HTTP_NOT_FOUND) {
+          result.success(null);
+        } else {
+          result.error(serviceException);
+        }
+      }
+    };
+  }
+
+  private RpcBatch.Callback<StorageObject> createUpdateCallback(final StorageOptions serviceOptions,
+      final StorageBatchResult<Blob> result) {
+    return new RpcBatch.Callback<StorageObject>() {
+      @Override
+      public void onSuccess(StorageObject response) {
+        result.success(response == null ? null : Blob.fromPb(serviceOptions.service(), response));
+      }
+
+      @Override
+      public void onFailure(GoogleJsonError googleJsonError) {
+        result.error(new StorageException(googleJsonError));
+      }
+    };
+  }
+}
diff --git a/gcloud-java-storage/src/main/java/com/google/cloud/storage/StorageBatchResult.java b/gcloud-java-storage/src/main/java/com/google/cloud/storage/StorageBatchResult.java
new file mode 100644
index 000000000000..2311436ab2d2
--- /dev/null
+++ b/gcloud-java-storage/src/main/java/com/google/cloud/storage/StorageBatchResult.java
@@ -0,0 +1,39 @@
+
+/*
+ * Copyright 2016 Google Inc. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.storage;
+
+import com.google.cloud.BatchResult;
+
+/**
+ * This class holds a single result of a batch call to Cloud Storage.
+ */
+public class StorageBatchResult<T> extends BatchResult<T, StorageException> {
+
+  StorageBatchResult() {
+  }
+
+  @Override
+  protected void error(StorageException error) {
+    super.error(error);
+  }
+
+  @Override
+  protected void success(T result) {
+    super.success(result);
+  }
+}
diff --git a/gcloud-java-storage/src/main/java/com/google/cloud/storage/StorageImpl.java b/gcloud-java-storage/src/main/java/com/google/cloud/storage/StorageImpl.java
index f4769905d8bc..1f72f1e3ad69 100644
--- a/gcloud-java-storage/src/main/java/com/google/cloud/storage/StorageImpl.java
+++ b/gcloud-java-storage/src/main/java/com/google/cloud/storage/StorageImpl.java
@@ -33,6 +33,7 @@
 
 import com.google.api.services.storage.model.StorageObject;
 import com.google.cloud.BaseService;
+import com.google.cloud.BatchResult;
 import com.google.cloud.Page;
 import com.google.cloud.PageImpl;
 import com.google.cloud.PageImpl.NextPageFetcher;
@@ -54,7 +55,6 @@
 
 import java.io.ByteArrayInputStream;
 import java.io.InputStream;
-import java.io.Serializable;
 import java.io.UnsupportedEncodingException;
 import java.net.MalformedURLException;
 import java.net.URL;
@@ -449,60 +449,8 @@ public byte[] call() {
   }
 
   @Override
-  public BatchResponse submit(BatchRequest batchRequest) {
-    List<Tuple<StorageObject, Map<StorageRpc.Option, ?>>> toDelete =
-        Lists.newArrayListWithCapacity(batchRequest.toDelete().size());
-    for (Map.Entry<BlobId, Iterable<BlobSourceOption>> entry : batchRequest.toDelete().entrySet()) {
-      BlobId blob = entry.getKey();
-      Map<StorageRpc.Option, ?> optionsMap = optionMap(blob.generation(), null, entry.getValue());
-      StorageObject storageObject = blob.toPb();
-      toDelete.add(Tuple.<StorageObject, Map<StorageRpc.Option, ?>>of(storageObject, optionsMap));
-    }
-    List<Tuple<StorageObject, Map<StorageRpc.Option, ?>>> toUpdate =
-        Lists.newArrayListWithCapacity(batchRequest.toUpdate().size());
-    for (Map.Entry<BlobInfo, Iterable<BlobTargetOption>> entry :
-        batchRequest.toUpdate().entrySet()) {
-      BlobInfo blobInfo = entry.getKey();
-      Map<StorageRpc.Option, ?> optionsMap =
-          optionMap(blobInfo.generation(), blobInfo.metageneration(), entry.getValue());
-      toUpdate.add(Tuple.<StorageObject, Map<StorageRpc.Option, ?>>of(blobInfo.toPb(), optionsMap));
-    }
-    List<Tuple<StorageObject, Map<StorageRpc.Option, ?>>> toGet =
-        Lists.newArrayListWithCapacity(batchRequest.toGet().size());
-    for (Map.Entry<BlobId, Iterable<BlobGetOption>> entry : batchRequest.toGet().entrySet()) {
-      BlobId blob = entry.getKey();
-      Map<StorageRpc.Option, ?> optionsMap = optionMap(blob.generation(), null, entry.getValue());
-      toGet.add(Tuple.<StorageObject, Map<StorageRpc.Option, ?>>of(blob.toPb(), optionsMap));
-    }
-    StorageRpc.BatchResponse response =
-        storageRpc.batch(new StorageRpc.BatchRequest(toDelete, toUpdate, toGet));
-    List<BatchResponse.Result<Boolean>> deletes =
-        transformBatchResult(toDelete, response.deletes, DELETE_FUNCTION);
-    List<BatchResponse.Result<Blob>> updates =
-        transformBatchResult(toUpdate, response.updates, Blob.BLOB_FROM_PB_FUNCTION);
-    List<BatchResponse.Result<Blob>> gets =
-        transformBatchResult(toGet, response.gets, Blob.BLOB_FROM_PB_FUNCTION);
-    return new BatchResponse(deletes, updates, gets);
-  }
-
-  private <I, O extends Serializable> List<BatchResponse.Result<O>> transformBatchResult(
-      Iterable<Tuple<StorageObject, Map<StorageRpc.Option, ?>>> request,
-      Map<StorageObject, Tuple<I, StorageException>> results,
-      Function<Tuple<Storage, I>, O> transform) {
-    List<BatchResponse.Result<O>> response = Lists.newArrayListWithCapacity(results.size());
-    for (Tuple<StorageObject, ?> tuple : request) {
-      Tuple<I, StorageException> result = results.get(tuple.x());
-      I object = result.x();
-      StorageException exception = result.y();
-      if (exception != null) {
-        response.add(new BatchResponse.Result<O>(exception));
-      } else {
-        response.add(object != null
-            ? BatchResponse.Result.of(transform.apply(Tuple.of((Storage) this, object)))
-            : BatchResponse.Result.<O>empty());
-      }
-    }
-    return response;
+  public StorageBatch batch() {
+    return new StorageBatch(this.options());
   }
 
   @Override
@@ -591,42 +539,80 @@ public URL signUrl(BlobInfo blobInfo, long duration, TimeUnit unit, SignUrlOptio
 
   @Override
   public List<Blob> get(BlobId... blobIds) {
-    BatchRequest.Builder requestBuilder = BatchRequest.builder();
+    return get(Arrays.asList(blobIds));
+  }
+
+  @Override
+  public List<Blob> get(Iterable<BlobId> blobIds) {
+    StorageBatch batch = batch();
+    final List<Blob> results = Lists.newArrayList();
     for (BlobId blob : blobIds) {
-      requestBuilder.get(blob);
+      batch.get(blob).notify(new BatchResult.Callback<Blob, StorageException>() {
+        @Override
+        public void success(Blob result) {
+          results.add(result);
+        }
+
+        @Override
+        public void error(StorageException exception) {
+          results.add(null);
+        }
+      });
     }
-    BatchResponse response = submit(requestBuilder.build());
-    return Collections.unmodifiableList(transformResultList(response.gets(), null));
+    batch.submit();
+    return Collections.unmodifiableList(results);
   }
 
   @Override
   public List<Blob> update(BlobInfo... blobInfos) {
-    BatchRequest.Builder requestBuilder = BatchRequest.builder();
+    return update(Arrays.asList(blobInfos));
+  }
+
+  @Override
+  public List<Blob> update(Iterable<BlobInfo> blobInfos) {
+    StorageBatch batch = batch();
+    final List<Blob> results = Lists.newArrayList();
     for (BlobInfo blobInfo : blobInfos) {
-      requestBuilder.update(blobInfo);
+      batch.update(blobInfo).notify(new BatchResult.Callback<Blob, StorageException>() {
+        @Override
+        public void success(Blob result) {
+          results.add(result);
+        }
+
+        @Override
+        public void error(StorageException exception) {
+          results.add(null);
+        }
+      });
     }
-    BatchResponse response = submit(requestBuilder.build());
-    return Collections.unmodifiableList(transformResultList(response.updates(), null));
+    batch.submit();
+    return Collections.unmodifiableList(results);
   }
 
   @Override
   public List<Boolean> delete(BlobId... blobIds) {
-    BatchRequest.Builder requestBuilder = BatchRequest.builder();
-    for (BlobId blob : blobIds) {
-      requestBuilder.delete(blob);
-    }
-    BatchResponse response = submit(requestBuilder.build());
-    return Collections.unmodifiableList(transformResultList(response.deletes(), Boolean.FALSE));
+    return delete(Arrays.asList(blobIds));
   }
 
-  private static <T extends Serializable> List<T> transformResultList(
-      List<BatchResponse.Result<T>> results, final T errorValue) {
-    return Lists.transform(results, new Function<BatchResponse.Result<T>, T>() {
-      @Override
-      public T apply(BatchResponse.Result<T> result) {
-        return result.failed() ? errorValue : result.get();
-      }
-    });
+  @Override
+  public List<Boolean> delete(Iterable<BlobId> blobIds) {
+    StorageBatch batch = batch();
+    final List<Boolean> results = Lists.newArrayList();
+    for (BlobId blob : blobIds) {
+      batch.delete(blob).notify(new BatchResult.Callback<Boolean, StorageException>() {
+        @Override
+        public void success(Boolean result) {
+          results.add(result);
+        }
+
+        @Override
+        public void error(StorageException exception) {
+          results.add(Boolean.FALSE);
+        }
+      });
+    }
+    batch.submit();
+    return Collections.unmodifiableList(results);
   }
 
   private static <T> void addToOptionMap(StorageRpc.Option option, T defaultValue,
@@ -646,12 +632,12 @@ private static <T> void addToOptionMap(StorageRpc.Option getOption, StorageRpc.O
     }
   }
 
-  private Map<StorageRpc.Option, ?> optionMap(Long generation, Long metaGeneration,
+  private static Map<StorageRpc.Option, ?> optionMap(Long generation, Long metaGeneration,
       Iterable<? extends Option> options) {
     return optionMap(generation, metaGeneration, options, false);
   }
 
-  private Map<StorageRpc.Option, ?> optionMap(Long generation, Long metaGeneration,
+  private static Map<StorageRpc.Option, ?> optionMap(Long generation, Long metaGeneration,
       Iterable<? extends Option> options, boolean useAsSource) {
     Map<StorageRpc.Option, Object> temp = Maps.newEnumMap(StorageRpc.Option.class);
     for (Option option : options) {
@@ -677,24 +663,24 @@ private static <T> void addToOptionMap(StorageRpc.Option getOption, StorageRpc.O
     return ImmutableMap.copyOf(temp);
   }
 
-  private Map<StorageRpc.Option, ?> optionMap(Option... options) {
+  private static Map<StorageRpc.Option, ?> optionMap(Option... options) {
     return optionMap(null, null, Arrays.asList(options));
   }
 
-  private Map<StorageRpc.Option, ?> optionMap(Long generation, Long metaGeneration,
+  private static Map<StorageRpc.Option, ?> optionMap(Long generation, Long metaGeneration,
       Option... options) {
     return optionMap(generation, metaGeneration, Arrays.asList(options));
   }
 
-  private Map<StorageRpc.Option, ?> optionMap(BucketInfo bucketInfo, Option... options) {
+  private static Map<StorageRpc.Option, ?> optionMap(BucketInfo bucketInfo, Option... options) {
     return optionMap(null, bucketInfo.metageneration(), options);
   }
 
-  private Map<StorageRpc.Option, ?> optionMap(BlobInfo blobInfo, Option... options) {
+  static Map<StorageRpc.Option, ?> optionMap(BlobInfo blobInfo, Option... options) {
     return optionMap(blobInfo.generation(), blobInfo.metageneration(), options);
   }
 
-  private Map<StorageRpc.Option, ?> optionMap(BlobId blobId, Option... options) {
+  static Map<StorageRpc.Option, ?> optionMap(BlobId blobId, Option... options) {
     return optionMap(blobId.generation(), null, options);
   }
 }
diff --git a/gcloud-java-storage/src/main/java/com/google/cloud/storage/spi/DefaultStorageRpc.java b/gcloud-java-storage/src/main/java/com/google/cloud/storage/spi/DefaultStorageRpc.java
index 4ef9afc45e4c..4d3c3775554e 100644
--- a/gcloud-java-storage/src/main/java/com/google/cloud/storage/spi/DefaultStorageRpc.java
+++ b/gcloud-java-storage/src/main/java/com/google/cloud/storage/spi/DefaultStorageRpc.java
@@ -34,6 +34,7 @@
 import static java.net.HttpURLConnection.HTTP_NOT_FOUND;
 import static javax.servlet.http.HttpServletResponse.SC_REQUESTED_RANGE_NOT_SATISFIABLE;
 
+import com.google.api.client.googleapis.batch.BatchRequest;
 import com.google.api.client.googleapis.batch.json.JsonBatchCallback;
 import com.google.api.client.googleapis.json.GoogleJsonError;
 import com.google.api.client.http.ByteArrayContent;
@@ -64,14 +65,13 @@
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Iterables;
 import com.google.common.collect.Lists;
-import com.google.common.collect.Maps;
 
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.math.BigInteger;
 import java.util.ArrayList;
-import java.util.Iterator;
+import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
 
@@ -82,7 +82,6 @@ public class DefaultStorageRpc implements StorageRpc {
   private final Storage storage;
 
   private static final long MEGABYTE = 1024L * 1024L;
-  private static final int MAX_BATCH_DELETES = 100;
 
   public DefaultStorageRpc(StorageOptions options) {
     HttpTransport transport = options.httpTransportFactory().create();
@@ -94,6 +93,82 @@ public DefaultStorageRpc(StorageOptions options) {
         .build();
   }
 
+  private class DefaultRpcBatch implements RpcBatch {
+
+    private static final int MAX_BATCH_DELETES = 100;
+
+    private final Storage storage;
+    private final LinkedList<BatchRequest> batches;
+    private int deleteCount;
+
+    private DefaultRpcBatch(Storage storage) {
+      this.storage = storage;
+      batches = new LinkedList<>();
+      batches.add(storage.batch());
+    }
+
+    @Override
+    public void addDelete(StorageObject storageObject, RpcBatch.Callback<Void> callback,
+        Map<Option, ?> options) {
+      try {
+        if (deleteCount == MAX_BATCH_DELETES) {
+          batches.add(storage.batch());
+          deleteCount = 0;
+        }
+        deleteCall(storageObject, options).queue(batches.getLast(), toJsonCallback(callback));
+        deleteCount++;
+      } catch (IOException ex) {
+        throw translate(ex);
+      }
+    }
+
+    @Override
+    public void addPatch(StorageObject storageObject, RpcBatch.Callback<StorageObject> callback,
+        Map<Option, ?> options) {
+      try {
+        patchCall(storageObject, options).queue(batches.getFirst(), toJsonCallback(callback));
+      } catch (IOException ex) {
+        throw translate(ex);
+      }
+    }
+
+    @Override
+    public void addGet(StorageObject storageObject, RpcBatch.Callback<StorageObject> callback,
+        Map<Option, ?> options) {
+      try {
+        getCall(storageObject, options).queue(batches.getFirst(), toJsonCallback(callback));
+      } catch (IOException ex) {
+        throw translate(ex);
+      }
+    }
+
+    @Override
+    public void submit() {
+      try {
+        for (BatchRequest batch : batches) {
+          batch.execute();
+        }
+      } catch (IOException ex) {
+        throw translate(ex);
+      }
+    }
+  }
+
+  private static <T> JsonBatchCallback<T> toJsonCallback(final RpcBatch.Callback<T> callback) {
+    return new JsonBatchCallback<T>() {
+      @Override
+      public void onSuccess(T response, HttpHeaders httpHeaders) throws IOException {
+        callback.onSuccess(response);
+      }
+
+      @Override
+      public void onFailure(GoogleJsonError googleJsonError, HttpHeaders httpHeaders)
+          throws IOException {
+        callback.onFailure(googleJsonError);
+      }
+    };
+  }
+
   private static StorageException translate(IOException exception) {
     return new StorageException(exception);
   }
@@ -209,20 +284,7 @@ public Bucket get(Bucket bucket, Map<Option, ?> options) {
     }
   }
 
-  @Override
-  public StorageObject get(StorageObject object, Map<Option, ?> options) {
-    try {
-      return getRequest(object, options).execute();
-    } catch (IOException ex) {
-      StorageException serviceException = translate(ex);
-      if (serviceException.code() == HTTP_NOT_FOUND) {
-        return null;
-      }
-      throw serviceException;
-    }
-  }
-
-  private Storage.Objects.Get getRequest(StorageObject object, Map<Option, ?> options)
+  private Storage.Objects.Get getCall(StorageObject object, Map<Option, ?> options)
       throws IOException {
     return storage.objects()
         .get(object.getBucket(), object.getName())
@@ -235,6 +297,19 @@ private Storage.Objects.Get getRequest(StorageObject object, Map<Option, ?> opti
         .setFields(FIELDS.getString(options));
   }
 
+  @Override
+  public StorageObject get(StorageObject object, Map<Option, ?> options) {
+    try {
+      return getCall(object, options).execute();
+    } catch (IOException ex) {
+      StorageException serviceException = translate(ex);
+      if (serviceException.code() == HTTP_NOT_FOUND) {
+        return null;
+      }
+      throw serviceException;
+    }
+  }
+
   @Override
   public Bucket patch(Bucket bucket, Map<Option, ?> options) {
     try {
@@ -251,16 +326,7 @@ public Bucket patch(Bucket bucket, Map<Option, ?> options) {
     }
   }
 
-  @Override
-  public StorageObject patch(StorageObject storageObject, Map<Option, ?> options) {
-    try {
-      return patchRequest(storageObject, options).execute();
-    } catch (IOException ex) {
-      throw translate(ex);
-    }
-  }
-
-  private Storage.Objects.Patch patchRequest(StorageObject storageObject, Map<Option, ?> options)
+  private Storage.Objects.Patch patchCall(StorageObject storageObject, Map<Option, ?> options)
       throws IOException {
     return storage.objects()
         .patch(storageObject.getBucket(), storageObject.getName(), storageObject)
@@ -272,6 +338,15 @@ private Storage.Objects.Patch patchRequest(StorageObject storageObject, Map<Opti
         .setIfGenerationNotMatch(IF_GENERATION_NOT_MATCH.getLong(options));
   }
 
+  @Override
+  public StorageObject patch(StorageObject storageObject, Map<Option, ?> options) {
+    try {
+      return patchCall(storageObject, options).execute();
+    } catch (IOException ex) {
+      throw translate(ex);
+    }
+  }
+
   @Override
   public boolean delete(Bucket bucket, Map<Option, ?> options) {
     try {
@@ -290,10 +365,21 @@ public boolean delete(Bucket bucket, Map<Option, ?> options) {
     }
   }
 
+  private Storage.Objects.Delete deleteCall(StorageObject blob, Map<Option, ?> options)
+      throws IOException {
+    return storage.objects()
+        .delete(blob.getBucket(), blob.getName())
+        .setGeneration(blob.getGeneration())
+        .setIfMetagenerationMatch(IF_METAGENERATION_MATCH.getLong(options))
+        .setIfMetagenerationNotMatch(IF_METAGENERATION_NOT_MATCH.getLong(options))
+        .setIfGenerationMatch(IF_GENERATION_MATCH.getLong(options))
+        .setIfGenerationNotMatch(IF_GENERATION_NOT_MATCH.getLong(options));
+  }
+
   @Override
   public boolean delete(StorageObject blob, Map<Option, ?> options) {
     try {
-      deleteRequest(blob, options).execute();
+      deleteCall(blob, options).execute();
       return true;
     } catch (IOException ex) {
       StorageException serviceException = translate(ex);
@@ -304,17 +390,6 @@ public boolean delete(StorageObject blob, Map<Option, ?> options) {
     }
   }
 
-  private Storage.Objects.Delete deleteRequest(StorageObject blob, Map<Option, ?> options)
-      throws IOException {
-    return storage.objects()
-        .delete(blob.getBucket(), blob.getName())
-        .setGeneration(blob.getGeneration())
-        .setIfMetagenerationMatch(IF_METAGENERATION_MATCH.getLong(options))
-        .setIfMetagenerationNotMatch(IF_METAGENERATION_NOT_MATCH.getLong(options))
-        .setIfGenerationMatch(IF_GENERATION_MATCH.getLong(options))
-        .setIfGenerationNotMatch(IF_GENERATION_NOT_MATCH.getLong(options));
-  }
-
   @Override
   public StorageObject compose(Iterable<StorageObject> sources, StorageObject target,
       Map<Option, ?> targetOptions) {
@@ -364,92 +439,8 @@ public byte[] load(StorageObject from, Map<Option, ?> options) {
   }
 
   @Override
-  public BatchResponse batch(BatchRequest request) {
-    List<List<Tuple<StorageObject, Map<Option, ?>>>> partitionedToDelete =
-        Lists.partition(request.toDelete, MAX_BATCH_DELETES);
-    Iterator<List<Tuple<StorageObject, Map<Option, ?>>>> iterator = partitionedToDelete.iterator();
-    BatchRequest chunkRequest = new BatchRequest(
-        iterator.hasNext()
-            ? iterator.next() : ImmutableList.<Tuple<StorageObject, Map<Option, ?>>>of(),
-        request.toUpdate, request.toGet);
-    BatchResponse response = batchChunk(chunkRequest);
-    Map<StorageObject, Tuple<Boolean, StorageException>> deletes =
-        Maps.newHashMapWithExpectedSize(request.toDelete.size());
-    deletes.putAll(response.deletes);
-    while (iterator.hasNext()) {
-      chunkRequest = new BatchRequest(iterator.next(), null, null);
-      BatchResponse deleteBatchResponse = batchChunk(chunkRequest);
-      deletes.putAll(deleteBatchResponse.deletes);
-    }
-    return new BatchResponse(deletes, response.updates, response.gets);
-  }
-
-  private BatchResponse batchChunk(BatchRequest request) {
-    com.google.api.client.googleapis.batch.BatchRequest batch = storage.batch();
-    final Map<StorageObject, Tuple<Boolean, StorageException>> deletes =
-        Maps.newConcurrentMap();
-    final Map<StorageObject, Tuple<StorageObject, StorageException>> updates =
-        Maps.newConcurrentMap();
-    final Map<StorageObject, Tuple<StorageObject, StorageException>> gets =
-        Maps.newConcurrentMap();
-    try {
-      for (final Tuple<StorageObject, Map<Option, ?>> tuple : request.toDelete) {
-        deleteRequest(tuple.x(), tuple.y()).queue(batch, new JsonBatchCallback<Void>() {
-          @Override
-          public void onSuccess(Void ignore, HttpHeaders responseHeaders) {
-            deletes.put(tuple.x(), Tuple.<Boolean, StorageException>of(Boolean.TRUE, null));
-          }
-
-          @Override
-          public void onFailure(GoogleJsonError e, HttpHeaders responseHeaders) {
-            if (e.getCode() == HTTP_NOT_FOUND) {
-              deletes.put(tuple.x(), Tuple.<Boolean, StorageException>of(Boolean.FALSE, null));
-            } else {
-              deletes.put(tuple.x(), Tuple.<Boolean, StorageException>of(null, translate(e)));
-            }
-          }
-        });
-      }
-      for (final Tuple<StorageObject, Map<Option, ?>> tuple : request.toUpdate) {
-        patchRequest(tuple.x(), tuple.y()).queue(batch, new JsonBatchCallback<StorageObject>() {
-          @Override
-          public void onSuccess(StorageObject storageObject, HttpHeaders responseHeaders) {
-            updates.put(tuple.x(),
-                Tuple.<StorageObject, StorageException>of(storageObject, null));
-          }
-
-          @Override
-          public void onFailure(GoogleJsonError e, HttpHeaders responseHeaders) {
-            updates.put(tuple.x(),
-                Tuple.<StorageObject, StorageException>of(null, translate(e)));
-          }
-        });
-      }
-      for (final Tuple<StorageObject, Map<Option, ?>> tuple : request.toGet) {
-        getRequest(tuple.x(), tuple.y()).queue(batch, new JsonBatchCallback<StorageObject>() {
-          @Override
-          public void onSuccess(StorageObject storageObject, HttpHeaders responseHeaders) {
-            gets.put(tuple.x(),
-                Tuple.<StorageObject, StorageException>of(storageObject, null));
-          }
-
-          @Override
-          public void onFailure(GoogleJsonError e, HttpHeaders responseHeaders) {
-            if (e.getCode() == HTTP_NOT_FOUND) {
-              gets.put(tuple.x(),
-                  Tuple.<StorageObject, StorageException>of(null, null));
-            } else {
-              gets.put(tuple.x(),
-                  Tuple.<StorageObject, StorageException>of(null, translate(e)));
-            }
-          }
-        });
-      }
-      batch.execute();
-    } catch (IOException ex) {
-      throw translate(ex);
-    }
-    return new BatchResponse(deletes, updates, gets);
+  public RpcBatch createBatch() {
+    return new DefaultRpcBatch(storage);
   }
 
   @Override
diff --git a/gcloud-java-storage/src/main/java/com/google/cloud/storage/spi/RpcBatch.java b/gcloud-java-storage/src/main/java/com/google/cloud/storage/spi/RpcBatch.java
new file mode 100644
index 000000000000..54b15fe45654
--- /dev/null
+++ b/gcloud-java-storage/src/main/java/com/google/cloud/storage/spi/RpcBatch.java
@@ -0,0 +1,70 @@
+/*
+ * Copyright 2016 Google Inc. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.storage.spi;
+
+import com.google.api.client.googleapis.json.GoogleJsonError;
+import com.google.api.services.storage.model.StorageObject;
+
+import java.util.Map;
+
+/**
+ * An interface for the collection of batch operations.
+ */
+public interface RpcBatch {
+
+  /**
+   * An interface for batch callbacks.
+   */
+  interface Callback<T> {
+
+    /**
+     * This method will be called upon success of the batch operation.
+     */
+    void onSuccess(T response);
+
+    /**
+     * This method will be called upon failure of the batch operation.
+     */
+    void onFailure(GoogleJsonError googleJsonError);
+  }
+
+  /**
+   * Adds a call to "delete storage object" to the batch, with the provided {@code callback} and
+   * {@code options}.
+   */
+  void addDelete(StorageObject storageObject, Callback<Void> callback,
+      Map<StorageRpc.Option, ?> options);
+
+  /**
+   * Adds a call to "patch storage object" to the batch, with the provided {@code callback} and
+   * {@code options}.
+   */
+  void addPatch(StorageObject storageObject, Callback<StorageObject> callback,
+      Map<StorageRpc.Option, ?> options);
+
+  /**
+   * Adds a call to "get storage object" to the batch, with the provided {@code callback} and
+   * {@code options}.
+   */
+  void addGet(StorageObject storageObject, Callback<StorageObject> callback,
+      Map<StorageRpc.Option, ?> options);
+
+  /**
+   * Submits a batch of requests for processing using a single RPC request to Cloud Storage.
+   */
+  void submit();
+}
diff --git a/gcloud-java-storage/src/main/java/com/google/cloud/storage/spi/StorageRpc.java b/gcloud-java-storage/src/main/java/com/google/cloud/storage/spi/StorageRpc.java
index 13dddb7d6737..6beeeed11f9f 100644
--- a/gcloud-java-storage/src/main/java/com/google/cloud/storage/spi/StorageRpc.java
+++ b/gcloud-java-storage/src/main/java/com/google/cloud/storage/spi/StorageRpc.java
@@ -16,16 +16,11 @@
 
 package com.google.cloud.storage.spi;
 
-import static com.google.common.base.MoreObjects.firstNonNull;
-
 import com.google.api.services.storage.model.Bucket;
 import com.google.api.services.storage.model.StorageObject;
 import com.google.cloud.storage.StorageException;
-import com.google.common.collect.ImmutableList;
-import com.google.common.collect.ImmutableMap;
 
 import java.io.InputStream;
-import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 
@@ -101,39 +96,6 @@ public Y y() {
     }
   }
 
-  class BatchRequest {
-
-    public final List<Tuple<StorageObject, Map<Option, ?>>> toDelete;
-    public final List<Tuple<StorageObject, Map<Option, ?>>> toUpdate;
-    public final List<Tuple<StorageObject, Map<Option, ?>>> toGet;
-
-    public BatchRequest(Iterable<Tuple<StorageObject, Map<Option, ?>>> toDelete,
-        Iterable<Tuple<StorageObject, Map<Option, ?>>> toUpdate,
-        Iterable<Tuple<StorageObject, Map<Option, ?>>> toGet) {
-      this.toDelete = ImmutableList.copyOf(
-          firstNonNull(toDelete, ImmutableList.<Tuple<StorageObject, Map<Option, ?>>>of()));
-      this.toUpdate = ImmutableList.copyOf(
-          firstNonNull(toUpdate, ImmutableList.<Tuple<StorageObject, Map<Option, ?>>>of()));
-      this.toGet = ImmutableList.copyOf(
-          firstNonNull(toGet, ImmutableList.<Tuple<StorageObject, Map<Option, ?>>>of()));
-    }
-  }
-
-  class BatchResponse {
-
-    public final Map<StorageObject, Tuple<Boolean, StorageException>> deletes;
-    public final Map<StorageObject, Tuple<StorageObject, StorageException>> updates;
-    public final Map<StorageObject, Tuple<StorageObject, StorageException>> gets;
-
-    public BatchResponse(Map<StorageObject, Tuple<Boolean, StorageException>> deletes,
-        Map<StorageObject, Tuple<StorageObject, StorageException>> updates,
-        Map<StorageObject, Tuple<StorageObject, StorageException>> gets) {
-      this.deletes = ImmutableMap.copyOf(deletes);
-      this.updates = ImmutableMap.copyOf(updates);
-      this.gets = ImmutableMap.copyOf(gets);
-    }
-  }
-
   class RewriteRequest {
 
     public final StorageObject source;
@@ -295,11 +257,9 @@ public int hashCode() {
   boolean delete(StorageObject object, Map<Option, ?> options);
 
   /**
-   * Sends a batch request.
-   *
-   * @throws StorageException upon failure
+   * Creates an empty batch.
    */
-  BatchResponse batch(BatchRequest request);
+  RpcBatch createBatch();
 
   /**
    * Sends a compose request.
diff --git a/gcloud-java-storage/src/test/java/com/google/cloud/storage/BatchRequestTest.java b/gcloud-java-storage/src/test/java/com/google/cloud/storage/BatchRequestTest.java
deleted file mode 100644
index a47c7c140216..000000000000
--- a/gcloud-java-storage/src/test/java/com/google/cloud/storage/BatchRequestTest.java
+++ /dev/null
@@ -1,142 +0,0 @@
-/*
- * Copyright 2015 Google Inc. All Rights Reserved.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *       http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.google.cloud.storage;
-
-import static com.google.cloud.storage.Storage.PredefinedAcl.PUBLIC_READ;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertNotEquals;
-import static org.junit.Assert.assertTrue;
-
-import com.google.cloud.storage.Storage.BlobGetOption;
-import com.google.cloud.storage.Storage.BlobSourceOption;
-import com.google.cloud.storage.Storage.BlobTargetOption;
-import com.google.common.collect.Iterables;
-
-import org.junit.Test;
-
-import java.util.Iterator;
-import java.util.Map.Entry;
-
-public class BatchRequestTest {
-
-  @Test
-  public void testBatchRequest() {
-    BatchRequest request = BatchRequest.builder()
-        .delete(BlobId.of("b1", "o1", 1L), BlobSourceOption.generationMatch())
-        .delete("b1", "o2", BlobSourceOption.generationMatch(1),
-            BlobSourceOption.metagenerationMatch(2))
-        .update(BlobInfo.builder("b2", "o1").build(), BlobTargetOption.predefinedAcl(PUBLIC_READ))
-        .update(BlobInfo.builder("b2", "o2").build())
-        .get(BlobId.of("b3", "o1", 1L), BlobGetOption.generationMatch())
-        .get("b3", "o2", BlobGetOption.generationMatch(1))
-        .get("b3", "o3")
-        .build();
-
-    Iterator<Entry<BlobId, Iterable<BlobSourceOption>>> deletes = request
-        .toDelete().entrySet().iterator();
-    Entry<BlobId, Iterable<BlobSourceOption>> delete = deletes.next();
-    assertEquals(BlobId.of("b1", "o1", 1L), delete.getKey());
-    assertEquals(1, Iterables.size(delete.getValue()));
-    assertEquals(BlobSourceOption.generationMatch(), Iterables.getFirst(delete.getValue(), null));
-    delete = deletes.next();
-    assertEquals(BlobId.of("b1", "o2"), delete.getKey());
-    assertEquals(2, Iterables.size(delete.getValue()));
-    assertEquals(BlobSourceOption.generationMatch(1L), Iterables.getFirst(delete.getValue(), null));
-    assertEquals(BlobSourceOption.metagenerationMatch(2L),
-        Iterables.get(delete.getValue(), 1, null));
-    assertFalse(deletes.hasNext());
-
-    Iterator<Entry<BlobInfo, Iterable<BlobTargetOption>>> updates = request
-        .toUpdate().entrySet().iterator();
-    Entry<BlobInfo, Iterable<BlobTargetOption>> update = updates.next();
-    assertEquals(BlobInfo.builder("b2", "o1").build(), update.getKey());
-    assertEquals(1, Iterables.size(update.getValue()));
-    assertEquals(BlobTargetOption.predefinedAcl(PUBLIC_READ),
-        Iterables.getFirst(update.getValue(), null));
-    update = updates.next();
-    assertEquals(BlobInfo.builder("b2", "o2").build(), update.getKey());
-    assertTrue(Iterables.isEmpty(update.getValue()));
-    assertFalse(updates.hasNext());
-
-    Iterator<Entry<BlobId, Iterable<BlobGetOption>>> gets = request.toGet().entrySet().iterator();
-    Entry<BlobId, Iterable<BlobGetOption>> get = gets.next();
-    assertEquals(BlobId.of("b3", "o1", 1L), get.getKey());
-    assertEquals(1, Iterables.size(get.getValue()));
-    assertEquals(BlobGetOption.generationMatch(), Iterables.getFirst(get.getValue(), null));
-    get = gets.next();
-    assertEquals(BlobId.of("b3", "o2"), get.getKey());
-    assertEquals(1, Iterables.size(get.getValue()));
-    assertEquals(BlobGetOption.generationMatch(1), Iterables.getFirst(get.getValue(), null));
-    get = gets.next();
-    assertEquals(BlobId.of("b3", "o3"), get.getKey());
-    assertTrue(Iterables.isEmpty(get.getValue()));
-    assertFalse(gets.hasNext());
-  }
-
-  @Test
-  public void testEquals() {
-    BatchRequest request = BatchRequest.builder()
-        .delete("b1", "o1")
-        .delete("b1", "o2")
-        .update(BlobInfo.builder("b2", "o1").build())
-        .update(BlobInfo.builder("b2", "o2").build())
-        .get("b3", "o1")
-        .get("b3", "o2")
-        .build();
-    BatchRequest requestEquals = BatchRequest.builder()
-        .delete("b1", "o1")
-        .delete("b1", "o2")
-        .update(BlobInfo.builder("b2", "o1").build())
-        .update(BlobInfo.builder("b2", "o2").build())
-        .get("b3", "o1")
-        .get("b3", "o2")
-        .build();
-    BatchRequest requestNotEquals1 = BatchRequest.builder()
-        .delete("b1", "o1")
-        .delete("b1", "o3")
-        .update(BlobInfo.builder("b2", "o1").build())
-        .update(BlobInfo.builder("b2", "o2").build())
-        .get("b3", "o1")
-        .get("b3", "o2")
-        .build();
-    BatchRequest requestNotEquals2 = BatchRequest.builder()
-        .delete("b1", "o1")
-        .delete("b1", "o2")
-        .update(BlobInfo.builder("b2", "o1").build())
-        .update(BlobInfo.builder("b2", "o3").build())
-        .get("b3", "o1")
-        .get("b3", "o2")
-        .build();
-    BatchRequest requestNotEquals3 = BatchRequest.builder()
-        .delete("b1", "o1")
-        .delete("b1", "o2")
-        .update(BlobInfo.builder("b2", "o1").build())
-        .update(BlobInfo.builder("b2", "o2").build())
-        .get("b3", "o1")
-        .get("b3", "o3")
-        .build();
-    assertEquals(request, requestEquals);
-    assertEquals(request.hashCode(), requestEquals.hashCode());
-    assertNotEquals(request, requestNotEquals1);
-    assertNotEquals(request.hashCode(), requestNotEquals1.hashCode());
-    assertNotEquals(request, requestNotEquals2);
-    assertNotEquals(request.hashCode(), requestNotEquals2.hashCode());
-    assertNotEquals(request, requestNotEquals3);
-    assertNotEquals(request.hashCode(), requestNotEquals3.hashCode());
-  }
-}
diff --git a/gcloud-java-storage/src/test/java/com/google/cloud/storage/BatchResponseTest.java b/gcloud-java-storage/src/test/java/com/google/cloud/storage/BatchResponseTest.java
deleted file mode 100644
index a9177489c2fd..000000000000
--- a/gcloud-java-storage/src/test/java/com/google/cloud/storage/BatchResponseTest.java
+++ /dev/null
@@ -1,84 +0,0 @@
-/*
- * Copyright 2015 Google Inc. All Rights Reserved.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *       http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.google.cloud.storage;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotEquals;
-
-import com.google.cloud.storage.BatchResponse.Result;
-import com.google.common.collect.ImmutableList;
-
-import org.easymock.EasyMock;
-import org.junit.Before;
-import org.junit.Test;
-
-import java.util.List;
-
-public class BatchResponseTest {
-
-  private Storage mockStorage;
-  private Blob blob1;
-  private Blob blob2;
-  private Blob blob3;
-
-  @Before
-  public void setUp() {
-    mockStorage = EasyMock.createMock(Storage.class);
-    EasyMock.expect(mockStorage.options()).andReturn(null).times(3);
-    EasyMock.replay(mockStorage);
-    blob1 = new Blob(mockStorage, new BlobInfo.BuilderImpl(BlobInfo.builder("b", "o1").build()));
-    blob2 = new Blob(mockStorage, new BlobInfo.BuilderImpl(BlobInfo.builder("b", "o2").build()));
-    blob3 = new Blob(mockStorage, new BlobInfo.BuilderImpl(BlobInfo.builder("b", "o3").build()));
-  }
-
-  @Test
-  public void testBatchResponse() {
-    List<Result<Boolean>> deletes = ImmutableList.of(Result.of(true), Result.of(false));
-    List<Result<Blob>> updates =
-        ImmutableList.of(Result.of(blob1), Result.of(blob2));
-    List<Result<Blob>> gets = ImmutableList.of(Result.of(blob2), Result.of(blob3));
-    BatchResponse response = new BatchResponse(deletes, updates, gets);
-    assertEquals(deletes, response.deletes());
-    assertEquals(updates, response.updates());
-    assertEquals(gets, response.gets());
-  }
-
-  @Test
-  public void testEquals() {
-    List<Result<Boolean>> deletes = ImmutableList.of(Result.of(true), Result.of(false));
-    List<Result<Blob>> updates =
-        ImmutableList.of(Result.of(blob1), Result.of(blob2));
-    List<Result<Blob>> gets = ImmutableList.of(Result.of(blob2), Result.of(blob3));
-    List<Result<Boolean>> otherDeletes = ImmutableList.of(Result.of(false), Result.of(true));
-    List<Result<Blob>> otherUpdates = ImmutableList.of(Result.of(blob2), Result.of(blob3));
-    List<Result<Blob>> otherGets =
-        ImmutableList.of(Result.of(blob1), Result.of(blob2));
-    BatchResponse response = new BatchResponse(deletes, updates, gets);
-    BatchResponse responseEquals = new BatchResponse(deletes, updates, gets);
-    BatchResponse responseNotEquals1 = new BatchResponse(otherDeletes, updates, gets);
-    BatchResponse responseNotEquals2 = new BatchResponse(deletes, otherUpdates, gets);
-    BatchResponse responseNotEquals3 = new BatchResponse(deletes, updates, otherGets);
-    assertEquals(response, responseEquals);
-    assertEquals(response.hashCode(), responseEquals.hashCode());
-    assertNotEquals(response, responseNotEquals1);
-    assertNotEquals(response.hashCode(), responseNotEquals1.hashCode());
-    assertNotEquals(response, responseNotEquals2);
-    assertNotEquals(response.hashCode(), responseNotEquals2.hashCode());
-    assertNotEquals(response, responseNotEquals3);
-    assertNotEquals(response.hashCode(), responseNotEquals3.hashCode());
-  }
-}
diff --git a/gcloud-java-storage/src/test/java/com/google/cloud/storage/BucketTest.java b/gcloud-java-storage/src/test/java/com/google/cloud/storage/BucketTest.java
index 0494dfcdf506..21e42d23c415 100644
--- a/gcloud-java-storage/src/test/java/com/google/cloud/storage/BucketTest.java
+++ b/gcloud-java-storage/src/test/java/com/google/cloud/storage/BucketTest.java
@@ -19,7 +19,6 @@
 import static com.google.cloud.storage.Acl.Project.ProjectRole.VIEWERS;
 import static com.google.cloud.storage.Acl.Role.READER;
 import static com.google.cloud.storage.Acl.Role.WRITER;
-import static org.easymock.EasyMock.capture;
 import static org.easymock.EasyMock.createMock;
 import static org.easymock.EasyMock.createStrictMock;
 import static org.easymock.EasyMock.expect;
@@ -34,12 +33,13 @@
 import com.google.cloud.PageImpl;
 import com.google.cloud.storage.Acl.Project;
 import com.google.cloud.storage.Acl.User;
-import com.google.cloud.storage.BatchResponse.Result;
 import com.google.cloud.storage.BucketInfo.AgeDeleteRule;
 import com.google.cloud.storage.BucketInfo.DeleteRule;
+import com.google.common.base.Function;
 import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Lists;
 
-import org.easymock.Capture;
+import org.easymock.EasyMock;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Rule;
@@ -50,9 +50,7 @@
 import java.io.InputStream;
 import java.util.Collections;
 import java.util.Iterator;
-import java.util.LinkedList;
 import java.util.List;
-import java.util.Set;
 
 public class BucketTest {
 
@@ -99,7 +97,7 @@ public class BucketTest {
   private StorageOptions mockOptions = createMock(StorageOptions.class);
   private Bucket bucket;
   private Bucket expectedBucket;
-  private Iterable<Blob> blobResults;
+  private List<Blob> blobResults;
 
   @Rule
   public ExpectedException thrown = ExpectedException.none();
@@ -249,33 +247,35 @@ public void testGet() throws Exception {
   }
 
   @Test
-  public void testGetAll() throws Exception {
+  public void testGetAllArray() throws Exception {
     initializeExpectedBucket(4);
-    Capture<BatchRequest> capturedBatchRequest = Capture.newInstance();
-    List<Result<Blob>> batchResultList = new LinkedList<>();
-    for (Blob info : blobResults) {
-      batchResultList.add(new Result<>(info));
-    }
-    BatchResponse response = new BatchResponse(Collections.<Result<Boolean>>emptyList(),
-        Collections.<Result<Blob>>emptyList(), batchResultList);
     expect(storage.options()).andReturn(mockOptions);
-    expect(storage.submit(capture(capturedBatchRequest))).andReturn(response);
-    expect(storage.options()).andReturn(mockOptions).times(3);
+    List<BlobId> blobIds = Lists.transform(blobResults, new Function<Blob, BlobId>() {
+      @Override
+      public BlobId apply(Blob blob) {
+        return blob.blobId();
+      }
+    });
+    expect(storage.get(blobIds)).andReturn(blobResults);
     replay(storage);
     initializeBucket();
-    List<Blob> blobs = bucket.get("n1", "n2", "n3");
-    Set<BlobId> blobInfoSet = capturedBatchRequest.getValue().toGet().keySet();
-    assertEquals(batchResultList.size(), blobInfoSet.size());
-    for (BlobInfo info : blobResults) {
-      assertTrue(blobInfoSet.contains(info.blobId()));
-    }
-    Iterator<Blob> blobIterator = blobs.iterator();
-    Iterator<Result<Blob>> batchResultIterator = response.gets().iterator();
-    while (batchResultIterator.hasNext() && blobIterator.hasNext()) {
-      assertEquals(batchResultIterator.next().get(), blobIterator.next());
-    }
-    assertFalse(batchResultIterator.hasNext());
-    assertFalse(blobIterator.hasNext());
+    assertEquals(blobResults, bucket.get("n1", "n2", "n3"));
+  }
+
+  @Test
+  public void testGetAllIterable() throws Exception {
+    initializeExpectedBucket(4);
+    expect(storage.options()).andReturn(mockOptions);
+    List<BlobId> blobIds = Lists.transform(blobResults, new Function<Blob, BlobId>() {
+      @Override
+      public BlobId apply(Blob blob) {
+        return blob.blobId();
+      }
+    });
+    expect(storage.get(blobIds)).andReturn(blobResults);
+    replay(storage);
+    initializeBucket();
+    assertEquals(blobResults, bucket.get(ImmutableList.of("n1", "n2", "n3")));
   }
 
   @Test
diff --git a/gcloud-java-storage/src/test/java/com/google/cloud/storage/SerializationTest.java b/gcloud-java-storage/src/test/java/com/google/cloud/storage/SerializationTest.java
index a8d399e1b1e4..a816415ddc02 100644
--- a/gcloud-java-storage/src/test/java/com/google/cloud/storage/SerializationTest.java
+++ b/gcloud-java-storage/src/test/java/com/google/cloud/storage/SerializationTest.java
@@ -45,11 +45,6 @@ public class SerializationTest extends BaseSerializationTest {
   private static final Cors.Origin ORIGIN = Cors.Origin.any();
   private static final Cors CORS =
       Cors.builder().maxAgeSeconds(1).origins(Collections.singleton(ORIGIN)).build();
-  private static final BatchRequest BATCH_REQUEST = BatchRequest.builder().delete("B", "N").build();
-  private static final BatchResponse BATCH_RESPONSE = new BatchResponse(
-      Collections.singletonList(BatchResponse.Result.of(true)),
-      Collections.<BatchResponse.Result<Blob>>emptyList(),
-      Collections.<BatchResponse.Result<Blob>>emptyList());
   private static final PageImpl<Blob> PAGE_RESULT =
       new PageImpl<>(null, "c", Collections.singletonList(BLOB));
   private static final StorageException STORAGE_EXCEPTION = new StorageException(42, "message");
@@ -78,10 +73,9 @@ protected Serializable[] serializableObjects() {
         .authCredentials(null)
         .build();
     return new Serializable[]{ACL_DOMAIN, ACL_GROUP, ACL_PROJECT_, ACL_USER, ACL_RAW, ACL,
-        BLOB_INFO, BLOB, BUCKET_INFO, BUCKET, ORIGIN, CORS, BATCH_REQUEST, BATCH_RESPONSE,
-        PAGE_RESULT, BLOB_LIST_OPTIONS, BLOB_SOURCE_OPTIONS, BLOB_TARGET_OPTIONS,
-        BUCKET_LIST_OPTIONS, BUCKET_SOURCE_OPTIONS, BUCKET_TARGET_OPTIONS, STORAGE_EXCEPTION,
-        options, otherOptions};
+        BLOB_INFO, BLOB, BUCKET_INFO, BUCKET, ORIGIN, CORS, PAGE_RESULT, BLOB_LIST_OPTIONS,
+        BLOB_SOURCE_OPTIONS, BLOB_TARGET_OPTIONS, BUCKET_LIST_OPTIONS, BUCKET_SOURCE_OPTIONS,
+        BUCKET_TARGET_OPTIONS, STORAGE_EXCEPTION, options, otherOptions};
   }
 
   @Override
diff --git a/gcloud-java-storage/src/test/java/com/google/cloud/storage/StorageBatchResultTest.java b/gcloud-java-storage/src/test/java/com/google/cloud/storage/StorageBatchResultTest.java
new file mode 100644
index 000000000000..cfa5f6fb9fc6
--- /dev/null
+++ b/gcloud-java-storage/src/test/java/com/google/cloud/storage/StorageBatchResultTest.java
@@ -0,0 +1,109 @@
+/*
+ * Copyright 2016 Google Inc. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.storage;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import com.google.cloud.BatchResult;
+
+import org.easymock.EasyMock;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.IOException;
+
+public class StorageBatchResultTest {
+
+  private StorageBatchResult<Boolean> result;
+
+  @Before
+  public void setUp() {
+    result = new StorageBatchResult<>();
+  }
+
+  @Test
+  public void testSuccess() {
+    assertFalse(result.completed());
+    try {
+      result.get();
+      fail("This was not completed yet.");
+    } catch (IllegalStateException ex) {
+      // expected
+    }
+    result.success(true);
+    assertTrue(result.get());
+  }
+
+  @Test
+  public void testError() {
+    assertFalse(result.completed());
+    try {
+      result.get();
+      fail("This was not completed yet.");
+    } catch (IllegalStateException ex) {
+      // expected
+    }
+    StorageException ex = new StorageException(new IOException("some error"));
+    result.error(ex);
+    try {
+      result.get();
+      fail("This is a failed operation and should have thrown a StorageException.");
+    } catch (StorageException real) {
+      assertSame(ex, real);
+    }
+  }
+
+  @Test
+  public void testNotifyError() {
+    StorageException ex = new StorageException(new IOException("some error"));
+    assertFalse(result.completed());
+    BatchResult.Callback<Boolean, StorageException> callback =
+        EasyMock.createStrictMock(BatchResult.Callback.class);
+    callback.error(ex);
+    EasyMock.replay(callback);
+    result.notify(callback);
+    result.error(ex);
+    try {
+      result.notify(callback);
+      fail("The batch has been completed.");
+    } catch (IllegalStateException exception) {
+      // expected
+    }
+    EasyMock.verify(callback);
+  }
+
+  @Test
+  public void testNotifySuccess() {
+    assertFalse(result.completed());
+    BatchResult.Callback<Boolean, StorageException> callback =
+        EasyMock.createStrictMock(BatchResult.Callback.class);
+    callback.success(true);
+    EasyMock.replay(callback);
+    result.notify(callback);
+    result.success(true);
+    try {
+      result.notify(callback);
+      fail("The batch has been completed.");
+    } catch (IllegalStateException exception) {
+      // expected
+    }
+    EasyMock.verify(callback);
+  }
+}
diff --git a/gcloud-java-storage/src/test/java/com/google/cloud/storage/StorageBatchTest.java b/gcloud-java-storage/src/test/java/com/google/cloud/storage/StorageBatchTest.java
new file mode 100644
index 000000000000..02f14fd457ed
--- /dev/null
+++ b/gcloud-java-storage/src/test/java/com/google/cloud/storage/StorageBatchTest.java
@@ -0,0 +1,224 @@
+/*
+ * Copyright 2016 Google Inc. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.storage;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import com.google.api.client.googleapis.json.GoogleJsonError;
+import com.google.api.services.storage.model.StorageObject;
+import com.google.cloud.storage.Storage.BlobGetOption;
+import com.google.cloud.storage.Storage.BlobSourceOption;
+import com.google.cloud.storage.Storage.BlobTargetOption;
+import com.google.cloud.storage.spi.RpcBatch;
+import com.google.cloud.storage.spi.StorageRpc;
+import com.google.common.collect.ImmutableMap;
+
+import org.easymock.Capture;
+import org.easymock.EasyMock;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.Map;
+
+public class StorageBatchTest {
+
+  private static final BlobId BLOB_ID = BlobId.of("b1", "n1");
+  private static final BlobId BLOB_ID_COMPLETE = BlobId.of("b1", "n1", 42L);
+  private static final BlobInfo BLOB_INFO = BlobInfo.builder(BLOB_ID).build();
+  private static final BlobInfo BLOB_INFO_COMPLETE = BlobInfo.builder(BLOB_ID_COMPLETE)
+      .metageneration(42L)
+      .build();
+  private static final BlobGetOption[] BLOB_GET_OPTIONS = {
+      BlobGetOption.generationMatch(42L), BlobGetOption.metagenerationMatch(42L)};
+  private static final BlobSourceOption[] BLOB_SOURCE_OPTIONS = {
+      BlobSourceOption.generationMatch(42L), BlobSourceOption.metagenerationMatch(42L)};
+  private static final BlobTargetOption[] BLOB_TARGET_OPTIONS = {
+      BlobTargetOption.generationMatch(), BlobTargetOption.metagenerationMatch()};
+  private static final GoogleJsonError GOOGLE_JSON_ERROR = new GoogleJsonError();
+
+  private StorageOptions optionsMock;
+  private StorageRpc dnsRpcMock;
+  private RpcBatch batchMock;
+  private StorageBatch dnsBatch;
+  private final Storage storage = EasyMock.createStrictMock(Storage.class);
+
+  @Before
+  public void setUp() {
+    optionsMock = EasyMock.createMock(StorageOptions.class);
+    dnsRpcMock = EasyMock.createMock(StorageRpc.class);
+    batchMock = EasyMock.createMock(RpcBatch.class);
+    EasyMock.expect(optionsMock.rpc()).andReturn(dnsRpcMock);
+    EasyMock.expect(dnsRpcMock.createBatch()).andReturn(batchMock);
+    EasyMock.replay(optionsMock, dnsRpcMock, batchMock, storage);
+    dnsBatch = new StorageBatch(optionsMock);
+  }
+
+  @After
+  public void tearDown() {
+    EasyMock.verify(batchMock, dnsRpcMock, optionsMock, storage);
+  }
+
+  @Test
+  public void testConstructor() {
+    assertSame(batchMock, dnsBatch.batch());
+    assertSame(optionsMock, dnsBatch.options());
+    assertSame(dnsRpcMock, dnsBatch.storageRpc());
+  }
+
+  @Test
+  public void testDelete() {
+    EasyMock.reset(batchMock);
+    Capture<RpcBatch.Callback<Void>> callback = Capture.newInstance();
+    batchMock.addDelete(EasyMock.eq(BLOB_INFO.toPb()), EasyMock.capture(callback),
+        EasyMock.eq(ImmutableMap.<StorageRpc.Option, Object>of()));
+    EasyMock.replay(batchMock);
+    StorageBatchResult<Boolean> batchResult = dnsBatch.delete(BLOB_ID.bucket(), BLOB_ID.name());
+    assertNotNull(callback.getValue());
+    try {
+      batchResult.get();
+      fail("No result available yet.");
+    } catch (IllegalStateException ex) {
+      // expected
+    }
+    // testing error here, success is tested with options
+    RpcBatch.Callback<Void> capturedCallback = callback.getValue();
+    capturedCallback.onFailure(GOOGLE_JSON_ERROR);
+    try {
+      batchResult.get();
+      fail("Should throw a StorageExcetion on error.");
+    } catch (StorageException ex) {
+      // expected
+    }
+  }
+
+  @Test
+  public void testDeleteWithOptions() {
+    EasyMock.reset(batchMock);
+    Capture<RpcBatch.Callback<Void>> callback = Capture.newInstance();
+    Capture<Map<StorageRpc.Option, Object>> capturedOptions = Capture.newInstance();
+    batchMock.addDelete(EasyMock.eq(BLOB_INFO.toPb()), EasyMock.capture(callback),
+        EasyMock.capture(capturedOptions));
+    EasyMock.replay(batchMock);
+    StorageBatchResult<Boolean> batchResult = dnsBatch.delete(BLOB_ID, BLOB_SOURCE_OPTIONS);
+    assertNotNull(callback.getValue());
+    assertEquals(2, capturedOptions.getValue().size());
+    for (BlobSourceOption option : BLOB_SOURCE_OPTIONS) {
+      assertEquals(option.value(), capturedOptions.getValue().get(option.rpcOption()));
+    }
+    RpcBatch.Callback<Void> capturedCallback = callback.getValue();
+    capturedCallback.onSuccess(null);
+    assertTrue(batchResult.get());
+  }
+
+  @Test
+  public void testUpdate() {
+    EasyMock.reset(batchMock);
+    Capture<RpcBatch.Callback<StorageObject>> callback = Capture.newInstance();
+    batchMock.addPatch(EasyMock.eq(BLOB_INFO.toPb()), EasyMock.capture(callback),
+        EasyMock.eq(ImmutableMap.<StorageRpc.Option, Object>of()));
+    EasyMock.replay(batchMock);
+    StorageBatchResult<Blob> batchResult = dnsBatch.update(BLOB_INFO);
+    assertNotNull(callback.getValue());
+    try {
+      batchResult.get();
+      fail("No result available yet.");
+    } catch (IllegalStateException ex) {
+      // expected
+    }
+    // testing error here, success is tested with options
+    RpcBatch.Callback<StorageObject> capturedCallback = callback.getValue();
+    capturedCallback.onFailure(GOOGLE_JSON_ERROR);
+    try {
+      batchResult.get();
+      fail("Should throw a StorageExcetion on error.");
+    } catch (StorageException ex) {
+      // expected
+    }
+  }
+
+  @Test
+  public void testUpdateWithOptions() {
+    EasyMock.reset(storage, batchMock, optionsMock);
+    EasyMock.expect(storage.options()).andReturn(optionsMock).times(2);
+    EasyMock.expect(optionsMock.service()).andReturn(storage);
+    Capture<RpcBatch.Callback<StorageObject>> callback = Capture.newInstance();
+    Capture<Map<StorageRpc.Option, Object>> capturedOptions = Capture.newInstance();
+    batchMock.addPatch(EasyMock.eq(BLOB_INFO_COMPLETE.toPb()), EasyMock.capture(callback),
+        EasyMock.capture(capturedOptions));
+    EasyMock.replay(batchMock, storage, optionsMock);
+    StorageBatchResult<Blob> batchResult = dnsBatch.update(BLOB_INFO_COMPLETE, BLOB_TARGET_OPTIONS);
+    assertNotNull(callback.getValue());
+    assertEquals(2, capturedOptions.getValue().size());
+    assertEquals(42L, capturedOptions.getValue().get(BLOB_TARGET_OPTIONS[0].rpcOption()));
+    assertEquals(42L, capturedOptions.getValue().get(BLOB_TARGET_OPTIONS[1].rpcOption()));
+    RpcBatch.Callback<StorageObject> capturedCallback = callback.getValue();
+    capturedCallback.onSuccess(BLOB_INFO.toPb());
+    assertEquals(new Blob(storage, new Blob.BuilderImpl(BLOB_INFO)), batchResult.get());
+  }
+
+  @Test
+  public void testGet() {
+    EasyMock.reset(batchMock);
+    Capture<RpcBatch.Callback<StorageObject>> callback = Capture.newInstance();
+    batchMock.addGet(EasyMock.eq(BLOB_INFO.toPb()), EasyMock.capture(callback),
+        EasyMock.eq(ImmutableMap.<StorageRpc.Option, Object>of()));
+    EasyMock.replay(batchMock);
+    StorageBatchResult<Blob> batchResult = dnsBatch.get(BLOB_ID.bucket(), BLOB_ID.name());
+    assertNotNull(callback.getValue());
+    try {
+      batchResult.get();
+      fail("No result available yet.");
+    } catch (IllegalStateException ex) {
+      // expected
+    }
+    // testing error here, success is tested with options
+    RpcBatch.Callback<StorageObject> capturedCallback = callback.getValue();
+    capturedCallback.onFailure(GOOGLE_JSON_ERROR);
+    try {
+      batchResult.get();
+      fail("Should throw a StorageExcetion on error.");
+    } catch (StorageException ex) {
+      // expected
+    }
+  }
+
+  @Test
+  public void testGetWithOptions() {
+    EasyMock.reset(storage, batchMock, optionsMock);
+    EasyMock.expect(storage.options()).andReturn(optionsMock).times(2);
+    EasyMock.expect(optionsMock.service()).andReturn(storage);
+    Capture<RpcBatch.Callback<StorageObject>> callback = Capture.newInstance();
+    Capture<Map<StorageRpc.Option, Object>> capturedOptions = Capture.newInstance();
+    batchMock.addGet(EasyMock.eq(BLOB_INFO.toPb()), EasyMock.capture(callback),
+        EasyMock.capture(capturedOptions));
+    EasyMock.replay(storage, batchMock, optionsMock);
+    StorageBatchResult<Blob> batchResult = dnsBatch.get(BLOB_ID, BLOB_GET_OPTIONS);
+    assertNotNull(callback.getValue());
+    assertEquals(2, capturedOptions.getValue().size());
+    for (BlobGetOption option : BLOB_GET_OPTIONS) {
+      assertEquals(option.value(), capturedOptions.getValue().get(option.rpcOption()));
+    }
+    RpcBatch.Callback<StorageObject> capturedCallback = callback.getValue();
+    capturedCallback.onSuccess(BLOB_INFO.toPb());
+    assertEquals(new Blob(storage, new Blob.BuilderImpl(BLOB_INFO)), batchResult.get());
+  }
+}
diff --git a/gcloud-java-storage/src/test/java/com/google/cloud/storage/StorageImplTest.java b/gcloud-java-storage/src/test/java/com/google/cloud/storage/StorageImplTest.java
index 47f776458876..74fe5044289b 100644
--- a/gcloud-java-storage/src/test/java/com/google/cloud/storage/StorageImplTest.java
+++ b/gcloud-java-storage/src/test/java/com/google/cloud/storage/StorageImplTest.java
@@ -19,11 +19,13 @@
 import static java.nio.charset.StandardCharsets.UTF_8;
 import static org.junit.Assert.assertArrayEquals;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertTrue;
 
+import com.google.api.client.googleapis.json.GoogleJsonError;
 import com.google.api.services.storage.model.StorageObject;
 import com.google.cloud.AuthCredentials.ServiceAccountAuthCredentials;
 import com.google.cloud.Page;
@@ -32,14 +34,13 @@
 import com.google.cloud.ServiceOptions;
 import com.google.cloud.WriteChannel;
 import com.google.cloud.storage.Storage.CopyRequest;
+import com.google.cloud.storage.spi.RpcBatch;
 import com.google.cloud.storage.spi.StorageRpc;
 import com.google.cloud.storage.spi.StorageRpc.Tuple;
 import com.google.cloud.storage.spi.StorageRpcFactory;
-import com.google.common.base.Function;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Iterables;
-import com.google.common.collect.Maps;
 import com.google.common.io.BaseEncoding;
 
 import org.easymock.Capture;
@@ -983,82 +984,16 @@ public void testReadAllBytesWithOptionsFromBlobId() {
   }
 
   @Test
-  public void testApply() {
-    BatchRequest req = BatchRequest.builder()
-        .delete(BUCKET_NAME1, BLOB_NAME1)
-        .update(BLOB_INFO2)
-        .get(BUCKET_NAME1, BLOB_NAME3)
-        .build();
-    List<StorageObject> toDelete = ImmutableList.of(BlobId.of(BUCKET_NAME1, BLOB_NAME1).toPb());
-    List<StorageObject> toUpdate = ImmutableList.of(BlobId.of(BUCKET_NAME1, BLOB_NAME2).toPb());
-    List<StorageObject> toGet = ImmutableList.of(BlobId.of(BUCKET_NAME1, BLOB_NAME3).toPb());
-    List<Map<StorageRpc.Option, ?>> deleteOptions =
-        ImmutableList.<Map<StorageRpc.Option, ?>>of(EMPTY_RPC_OPTIONS);
-    List<Map<StorageRpc.Option, ?>> updateOptions =
-        ImmutableList.<Map<StorageRpc.Option, ?>>of(EMPTY_RPC_OPTIONS);
-    List<Map<StorageRpc.Option, ?>> getOptions =
-        ImmutableList.<Map<StorageRpc.Option, ?>>of(EMPTY_RPC_OPTIONS);
-
-    Map<StorageObject, Tuple<Boolean, StorageException>> deleteResult = Maps.toMap(toDelete,
-        new Function<StorageObject, Tuple<Boolean, StorageException>>() {
-          @Override
-          public Tuple<Boolean, StorageException> apply(StorageObject f) {
-            return Tuple.of(true, null);
-          }
-        });
-    Map<StorageObject, Tuple<StorageObject, StorageException>> updateResult = Maps.toMap(toUpdate,
-        new Function<StorageObject, Tuple<StorageObject, StorageException>>() {
-          @Override
-          public Tuple<StorageObject, StorageException> apply(StorageObject f) {
-            return Tuple.of(f, null);
-          }
-        });
-    Map<StorageObject, Tuple<StorageObject, StorageException>> getResult = Maps.toMap(toGet,
-        new Function<StorageObject, Tuple<StorageObject, StorageException>>() {
-          @Override
-          public Tuple<StorageObject, StorageException> apply(StorageObject f) {
-            return Tuple.of(f, null);
-          }
-        });
-    StorageRpc.BatchResponse res =
-        new StorageRpc.BatchResponse(deleteResult, updateResult, getResult);
-
-    Capture<StorageRpc.BatchRequest> capturedBatchRequest = Capture.newInstance();
-    EasyMock.expect(storageRpcMock.batch(EasyMock.capture(capturedBatchRequest))).andReturn(res);
-    EasyMock.replay(storageRpcMock);
+  public void testBatch() {
+    RpcBatch batchMock = EasyMock.mock(RpcBatch.class);
+    EasyMock.expect(storageRpcMock.createBatch()).andReturn(batchMock);
+    EasyMock.replay(batchMock, storageRpcMock);
     initializeService();
-    BatchResponse batchResponse = storage.submit(req);
-
-    // Verify captured StorageRpc.BatchRequest
-    List<Tuple<StorageObject, Map<StorageRpc.Option, ?>>> capturedToDelete =
-        capturedBatchRequest.getValue().toDelete;
-    List<Tuple<StorageObject, Map<StorageRpc.Option, ?>>> capturedToUpdate =
-        capturedBatchRequest.getValue().toUpdate;
-    List<Tuple<StorageObject, Map<StorageRpc.Option, ?>>> capturedToGet =
-        capturedBatchRequest.getValue().toGet;
-    for (int i = 0; i < capturedToDelete.size(); i++) {
-      assertEquals(toDelete.get(i), capturedToDelete.get(i).x());
-      assertEquals(deleteOptions.get(i), capturedToDelete.get(i).y());
-    }
-    for (int i = 0; i < capturedToUpdate.size(); i++) {
-      assertEquals(toUpdate.get(i), capturedToUpdate.get(i).x());
-      assertEquals(updateOptions.get(i), capturedToUpdate.get(i).y());
-    }
-    for (int i = 0; i < capturedToGet.size(); i++) {
-      assertEquals(toGet.get(i), capturedToGet.get(i).x());
-      assertEquals(getOptions.get(i), capturedToGet.get(i).y());
-    }
-
-    // Verify BatchResponse
-    for (BatchResponse.Result<Boolean> result : batchResponse.deletes()) {
-      assertTrue(result.get());
-    }
-    for (int i = 0; i < batchResponse.updates().size(); i++) {
-      assertEquals(toUpdate.get(i), batchResponse.updates().get(i).get().toPb());
-    }
-    for (int i = 0; i < batchResponse.gets().size(); i++) {
-      assertEquals(toGet.get(i), batchResponse.gets().get(i).get().toPb());
-    }
+    StorageBatch batch = storage.batch();
+    assertSame(options, batch.options());
+    assertSame(storageRpcMock, batch.storageRpc());
+    assertSame(batchMock, batch.batch());
+    EasyMock.verify(batchMock);
   }
 
   @Test
@@ -1187,129 +1122,143 @@ public void testSignUrlWithOptions() throws NoSuchAlgorithmException, InvalidKey
   }
 
   @Test
-  public void testGetAll() {
+  public void testGetAllArray() {
     BlobId blobId1 = BlobId.of(BUCKET_NAME1, BLOB_NAME1);
     BlobId blobId2 = BlobId.of(BUCKET_NAME1, BLOB_NAME2);
-    StorageObject storageObject1 = blobId1.toPb();
-    StorageObject storageObject2 = blobId2.toPb();
-    List<StorageObject> toGet = ImmutableList.of(storageObject1, storageObject2);
-
-    Map<StorageObject, Tuple<Boolean, StorageException>> deleteResult = ImmutableMap.of();
-    Map<StorageObject, Tuple<StorageObject, StorageException>> updateResult = ImmutableMap.of();
-    Map<StorageObject, Tuple<StorageObject, StorageException>> getResult = Maps.toMap(toGet,
-        new Function<StorageObject, Tuple<StorageObject, StorageException>>() {
-          @Override
-          public Tuple<StorageObject, StorageException> apply(StorageObject f) {
-            return Tuple.of(f, null);
-          }
-        });
-    StorageRpc.BatchResponse res =
-        new StorageRpc.BatchResponse(deleteResult, updateResult, getResult);
-
-    Capture<StorageRpc.BatchRequest> capturedBatchRequest = Capture.newInstance();
-    EasyMock.expect(storageRpcMock.batch(EasyMock.capture(capturedBatchRequest))).andReturn(res);
-    EasyMock.replay(storageRpcMock);
+    RpcBatch batchMock = EasyMock.createMock(RpcBatch.class);
+    Capture<RpcBatch.Callback<StorageObject>> callback1 = Capture.newInstance();
+    Capture<RpcBatch.Callback<StorageObject>> callback2 = Capture.newInstance();
+    batchMock.addGet(EasyMock.eq(blobId1.toPb()), EasyMock.capture(callback1),
+        EasyMock.eq(ImmutableMap.<StorageRpc.Option, Object>of()));
+    batchMock.addGet(EasyMock.eq(blobId2.toPb()), EasyMock.capture(callback2),
+        EasyMock.eq(ImmutableMap.<StorageRpc.Option, Object>of()));
+    EasyMock.expect(storageRpcMock.createBatch()).andReturn(batchMock);
+    batchMock.submit();
+    EasyMock.replay(storageRpcMock, batchMock);
     initializeService();
     List<Blob> resultBlobs = storage.get(blobId1, blobId2);
+    callback1.getValue().onSuccess(BLOB_INFO1.toPb());
+    callback2.getValue().onFailure(new GoogleJsonError());
+    assertEquals(2, resultBlobs.size());
+    assertEquals(new Blob(storage, new BlobInfo.BuilderImpl(BLOB_INFO1)), resultBlobs.get(0));
+    assertNull(resultBlobs.get(1));
+    EasyMock.verify(batchMock);
+  }
 
-    // Verify captured StorageRpc.BatchRequest
-    List<Tuple<StorageObject, Map<StorageRpc.Option, ?>>> capturedToGet =
-        capturedBatchRequest.getValue().toGet;
-    assertTrue(capturedBatchRequest.getValue().toDelete.isEmpty());
-    assertTrue(capturedBatchRequest.getValue().toUpdate.isEmpty());
-    for (int i = 0; i < capturedToGet.size(); i++) {
-      assertEquals(toGet.get(i), capturedToGet.get(i).x());
-      assertTrue(capturedToGet.get(i).y().isEmpty());
-    }
-
-    // Verify result
-    for (int i = 0; i < resultBlobs.size(); i++) {
-      assertEquals(toGet.get(i), resultBlobs.get(i).toPb());
-    }
+  @Test
+  public void testGetAllArrayIterable() {
+    BlobId blobId1 = BlobId.of(BUCKET_NAME1, BLOB_NAME1);
+    BlobId blobId2 = BlobId.of(BUCKET_NAME1, BLOB_NAME2);
+    RpcBatch batchMock = EasyMock.createMock(RpcBatch.class);
+    Capture<RpcBatch.Callback<StorageObject>> callback1 = Capture.newInstance();
+    Capture<RpcBatch.Callback<StorageObject>> callback2 = Capture.newInstance();
+    batchMock.addGet(EasyMock.eq(blobId1.toPb()), EasyMock.capture(callback1),
+        EasyMock.eq(ImmutableMap.<StorageRpc.Option, Object>of()));
+    batchMock.addGet(EasyMock.eq(blobId2.toPb()), EasyMock.capture(callback2),
+        EasyMock.eq(ImmutableMap.<StorageRpc.Option, Object>of()));
+    EasyMock.expect(storageRpcMock.createBatch()).andReturn(batchMock);
+    batchMock.submit();
+    EasyMock.replay(storageRpcMock, batchMock);
+    initializeService();
+    List<Blob> resultBlobs = storage.get(ImmutableList.of(blobId1, blobId2));
+    callback1.getValue().onSuccess(BLOB_INFO1.toPb());
+    callback2.getValue().onFailure(new GoogleJsonError());
+    assertEquals(2, resultBlobs.size());
+    assertEquals(new Blob(storage, new BlobInfo.BuilderImpl(BLOB_INFO1)), resultBlobs.get(0));
+    assertNull(resultBlobs.get(1));
+    EasyMock.verify(batchMock);
   }
 
   @Test
-  public void testUpdateAll() {
-    BlobInfo blobInfo1 = BlobInfo.builder(BUCKET_NAME1, BLOB_NAME1).contentType("type").build();
-    BlobInfo blobInfo2 = BlobInfo.builder(BUCKET_NAME1, BLOB_NAME2).contentType("type").build();
-    StorageObject storageObject1 = blobInfo1.toPb();
-    StorageObject storageObject2 = blobInfo2.toPb();
-    List<StorageObject> toUpdate = ImmutableList.of(storageObject1, storageObject2);
-
-    Map<StorageObject, Tuple<Boolean, StorageException>> deleteResult = ImmutableMap.of();
-    Map<StorageObject, Tuple<StorageObject, StorageException>> getResult = ImmutableMap.of();
-    Map<StorageObject, Tuple<StorageObject, StorageException>> updateResult = Maps.toMap(toUpdate,
-        new Function<StorageObject, Tuple<StorageObject, StorageException>>() {
-          @Override
-          public Tuple<StorageObject, StorageException> apply(StorageObject f) {
-            return Tuple.of(f, null);
-          }
-        });
-    StorageRpc.BatchResponse res =
-        new StorageRpc.BatchResponse(deleteResult, updateResult, getResult);
-
-    Capture<StorageRpc.BatchRequest> capturedBatchRequest = Capture.newInstance();
-    EasyMock.expect(storageRpcMock.batch(EasyMock.capture(capturedBatchRequest))).andReturn(res);
-    EasyMock.replay(storageRpcMock);
+  public void testDeleteAllArray() {
+    BlobId blobId1 = BlobId.of(BUCKET_NAME1, BLOB_NAME1);
+    BlobId blobId2 = BlobId.of(BUCKET_NAME1, BLOB_NAME2);
+    RpcBatch batchMock = EasyMock.createMock(RpcBatch.class);
+    Capture<RpcBatch.Callback<Void>> callback1 = Capture.newInstance();
+    Capture<RpcBatch.Callback<Void>> callback2 = Capture.newInstance();
+    batchMock.addDelete(EasyMock.eq(blobId1.toPb()), EasyMock.capture(callback1),
+        EasyMock.eq(ImmutableMap.<StorageRpc.Option, Object>of()));
+    batchMock.addDelete(EasyMock.eq(blobId2.toPb()), EasyMock.capture(callback2),
+        EasyMock.eq(ImmutableMap.<StorageRpc.Option, Object>of()));
+    EasyMock.expect(storageRpcMock.createBatch()).andReturn(batchMock);
+    batchMock.submit();
+    EasyMock.replay(storageRpcMock, batchMock);
     initializeService();
-    List<Blob> resultBlobs = storage.update(blobInfo1, blobInfo2);
-
-    // Verify captured StorageRpc.BatchRequest
-    List<Tuple<StorageObject, Map<StorageRpc.Option, ?>>> capturedToUpdate =
-        capturedBatchRequest.getValue().toUpdate;
-    assertTrue(capturedBatchRequest.getValue().toDelete.isEmpty());
-    assertTrue(capturedBatchRequest.getValue().toGet.isEmpty());
-    for (int i = 0; i < capturedToUpdate.size(); i++) {
-      assertEquals(toUpdate.get(i), capturedToUpdate.get(i).x());
-      assertTrue(capturedToUpdate.get(i).y().isEmpty());
-    }
+    List<Boolean> result = storage.delete(blobId1, blobId2);
+    callback1.getValue().onSuccess(null);
+    callback2.getValue().onFailure(new GoogleJsonError());
+    assertEquals(2, result.size());
+    assertTrue(result.get(0));
+    assertFalse(result.get(1));
+    EasyMock.verify(batchMock);
+  }
 
-    // Verify result
-    for (int i = 0; i < resultBlobs.size(); i++) {
-      assertEquals(toUpdate.get(i), resultBlobs.get(i).toPb());
-    }
+  @Test
+  public void testDeleteAllIterable() {
+    BlobId blobId1 = BlobId.of(BUCKET_NAME1, BLOB_NAME1);
+    BlobId blobId2 = BlobId.of(BUCKET_NAME1, BLOB_NAME2);
+    RpcBatch batchMock = EasyMock.createMock(RpcBatch.class);
+    Capture<RpcBatch.Callback<Void>> callback1 = Capture.newInstance();
+    Capture<RpcBatch.Callback<Void>> callback2 = Capture.newInstance();
+    batchMock.addDelete(EasyMock.eq(blobId1.toPb()), EasyMock.capture(callback1),
+        EasyMock.eq(ImmutableMap.<StorageRpc.Option, Object>of()));
+    batchMock.addDelete(EasyMock.eq(blobId2.toPb()), EasyMock.capture(callback2),
+        EasyMock.eq(ImmutableMap.<StorageRpc.Option, Object>of()));
+    EasyMock.expect(storageRpcMock.createBatch()).andReturn(batchMock);
+    batchMock.submit();
+    EasyMock.replay(storageRpcMock, batchMock);
+    initializeService();
+    List<Boolean> result = storage.delete(blobId1, blobId2);
+    callback1.getValue().onSuccess(null);
+    callback2.getValue().onFailure(new GoogleJsonError());
+    assertEquals(2, result.size());
+    assertTrue(result.get(0));
+    assertFalse(result.get(1));
+    EasyMock.verify(batchMock);
   }
 
   @Test
-  public void testDeleteAll() {
-    BlobInfo blobInfo1 = BlobInfo.builder(BUCKET_NAME1, BLOB_NAME1).build();
-    BlobInfo blobInfo2 = BlobInfo.builder(BUCKET_NAME1, BLOB_NAME2).build();
-    StorageObject storageObject1 = blobInfo1.toPb();
-    StorageObject storageObject2 = blobInfo2.toPb();
-    List<StorageObject> toUpdate = ImmutableList.of(storageObject1, storageObject2);
-
-    Map<StorageObject, Tuple<StorageObject, StorageException>> updateResult = ImmutableMap.of();
-    Map<StorageObject, Tuple<StorageObject, StorageException>> getResult = ImmutableMap.of();
-    Map<StorageObject, Tuple<Boolean, StorageException>> deleteResult = Maps.toMap(toUpdate,
-        new Function<StorageObject, Tuple<Boolean, StorageException>>() {
-          @Override
-          public Tuple<Boolean, StorageException> apply(StorageObject f) {
-            return Tuple.of(true, null);
-          }
-        });
-    StorageRpc.BatchResponse res =
-        new StorageRpc.BatchResponse(deleteResult, updateResult, getResult);
-
-    Capture<StorageRpc.BatchRequest> capturedBatchRequest = Capture.newInstance();
-    EasyMock.expect(storageRpcMock.batch(EasyMock.capture(capturedBatchRequest))).andReturn(res);
-    EasyMock.replay(storageRpcMock);
+  public void testUpdateAllArray() {
+    RpcBatch batchMock = EasyMock.createMock(RpcBatch.class);
+    Capture<RpcBatch.Callback<StorageObject>> callback1 = Capture.newInstance();
+    Capture<RpcBatch.Callback<StorageObject>> callback2 = Capture.newInstance();
+    batchMock.addPatch(EasyMock.eq(BLOB_INFO1.toPb()), EasyMock.capture(callback1),
+        EasyMock.eq(ImmutableMap.<StorageRpc.Option, Object>of()));
+    batchMock.addPatch(EasyMock.eq(BLOB_INFO2.toPb()), EasyMock.capture(callback2),
+        EasyMock.eq(ImmutableMap.<StorageRpc.Option, Object>of()));
+    EasyMock.expect(storageRpcMock.createBatch()).andReturn(batchMock);
+    batchMock.submit();
+    EasyMock.replay(storageRpcMock, batchMock);
     initializeService();
-    List<Boolean> deleteResults = storage.delete(blobInfo1.blobId(), blobInfo2.blobId());
-
-    // Verify captured StorageRpc.BatchRequest
-    List<Tuple<StorageObject, Map<StorageRpc.Option, ?>>> capturedToDelete =
-        capturedBatchRequest.getValue().toDelete;
-    assertTrue(capturedBatchRequest.getValue().toUpdate.isEmpty());
-    assertTrue(capturedBatchRequest.getValue().toGet.isEmpty());
-    for (int i = 0; i < capturedToDelete.size(); i++) {
-      assertEquals(toUpdate.get(i), capturedToDelete.get(i).x());
-      assertTrue(capturedToDelete.get(i).y().isEmpty());
-    }
+    List<Blob> resultBlobs = storage.update(BLOB_INFO1, BLOB_INFO2);
+    callback1.getValue().onSuccess(BLOB_INFO1.toPb());
+    callback2.getValue().onFailure(new GoogleJsonError());
+    assertEquals(2, resultBlobs.size());
+    assertEquals(new Blob(storage, new BlobInfo.BuilderImpl(BLOB_INFO1)), resultBlobs.get(0));
+    assertNull(resultBlobs.get(1));
+    EasyMock.verify(batchMock);
+  }
 
-    // Verify result
-    for (Boolean deleteStatus : deleteResults) {
-      assertTrue(deleteStatus);
-    }
+  @Test
+  public void testUpdateAllIterable() {
+    RpcBatch batchMock = EasyMock.createMock(RpcBatch.class);
+    Capture<RpcBatch.Callback<StorageObject>> callback1 = Capture.newInstance();
+    Capture<RpcBatch.Callback<StorageObject>> callback2 = Capture.newInstance();
+    batchMock.addPatch(EasyMock.eq(BLOB_INFO1.toPb()), EasyMock.capture(callback1),
+        EasyMock.eq(ImmutableMap.<StorageRpc.Option, Object>of()));
+    batchMock.addPatch(EasyMock.eq(BLOB_INFO2.toPb()), EasyMock.capture(callback2),
+        EasyMock.eq(ImmutableMap.<StorageRpc.Option, Object>of()));
+    EasyMock.expect(storageRpcMock.createBatch()).andReturn(batchMock);
+    batchMock.submit();
+    EasyMock.replay(storageRpcMock, batchMock);
+    initializeService();
+    List<Blob> resultBlobs = storage.update(ImmutableList.of(BLOB_INFO1, BLOB_INFO2));
+    callback1.getValue().onSuccess(BLOB_INFO1.toPb());
+    callback2.getValue().onFailure(new GoogleJsonError());
+    assertEquals(2, resultBlobs.size());
+    assertEquals(new Blob(storage, new BlobInfo.BuilderImpl(BLOB_INFO1)), resultBlobs.get(0));
+    assertNull(resultBlobs.get(1));
+    EasyMock.verify(batchMock);
   }
 
   @Test
diff --git a/gcloud-java-storage/src/test/java/com/google/cloud/storage/it/ITStorageTest.java b/gcloud-java-storage/src/test/java/com/google/cloud/storage/it/ITStorageTest.java
index 684653390d1b..24c8fb1579d2 100644
--- a/gcloud-java-storage/src/test/java/com/google/cloud/storage/it/ITStorageTest.java
+++ b/gcloud-java-storage/src/test/java/com/google/cloud/storage/it/ITStorageTest.java
@@ -29,8 +29,6 @@
 import com.google.cloud.ReadChannel;
 import com.google.cloud.RestorableState;
 import com.google.cloud.WriteChannel;
-import com.google.cloud.storage.BatchRequest;
-import com.google.cloud.storage.BatchResponse;
 import com.google.cloud.storage.Blob;
 import com.google.cloud.storage.BlobId;
 import com.google.cloud.storage.BlobInfo;
@@ -41,6 +39,8 @@
 import com.google.cloud.storage.Storage;
 import com.google.cloud.storage.Storage.BlobField;
 import com.google.cloud.storage.Storage.BucketField;
+import com.google.cloud.storage.StorageBatch;
+import com.google.cloud.storage.StorageBatchResult;
 import com.google.cloud.storage.StorageException;
 import com.google.cloud.storage.testing.RemoteStorageHelper;
 import com.google.common.collect.ImmutableList;
@@ -778,16 +778,12 @@ public void testBatchRequest() {
     // Batch update request
     BlobInfo updatedBlob1 = sourceBlob1.toBuilder().contentType(CONTENT_TYPE).build();
     BlobInfo updatedBlob2 = sourceBlob2.toBuilder().contentType(CONTENT_TYPE).build();
-    BatchRequest updateRequest = BatchRequest.builder()
-        .update(updatedBlob1)
-        .update(updatedBlob2)
-        .build();
-    BatchResponse updateResponse = storage.submit(updateRequest);
-    assertEquals(2, updateResponse.updates().size());
-    assertEquals(0, updateResponse.deletes().size());
-    assertEquals(0, updateResponse.gets().size());
-    BlobInfo remoteUpdatedBlob1 = updateResponse.updates().get(0).get();
-    BlobInfo remoteUpdatedBlob2 = updateResponse.updates().get(1).get();
+    StorageBatch updateBatch = storage.batch();
+    StorageBatchResult<Blob> updateResult1 = updateBatch.update(updatedBlob1);
+    StorageBatchResult<Blob> updateResult2 = updateBatch.update(updatedBlob2);
+    updateBatch.submit();
+    Blob remoteUpdatedBlob1 = updateResult1.get();
+    Blob remoteUpdatedBlob2 = updateResult2.get();
     assertEquals(sourceBlob1.bucket(), remoteUpdatedBlob1.bucket());
     assertEquals(sourceBlob1.name(), remoteUpdatedBlob1.name());
     assertEquals(sourceBlob2.bucket(), remoteUpdatedBlob2.bucket());
@@ -796,41 +792,35 @@ public void testBatchRequest() {
     assertEquals(updatedBlob2.contentType(), remoteUpdatedBlob2.contentType());
 
     // Batch get request
-    BatchRequest getRequest = BatchRequest.builder()
-        .get(BUCKET, sourceBlobName1)
-        .get(BUCKET, sourceBlobName2)
-        .build();
-    BatchResponse getResponse = storage.submit(getRequest);
-    assertEquals(2, getResponse.gets().size());
-    assertEquals(0, getResponse.deletes().size());
-    assertEquals(0, getResponse.updates().size());
-    BlobInfo remoteBlob1 = getResponse.gets().get(0).get();
-    BlobInfo remoteBlob2 = getResponse.gets().get(1).get();
+    StorageBatch getBatch = storage.batch();
+    StorageBatchResult<Blob> getResult1 = getBatch.get(BUCKET, sourceBlobName1);
+    StorageBatchResult<Blob> getResult2 = getBatch.get(BUCKET, sourceBlobName2);
+    getBatch.submit();
+    Blob remoteBlob1 = getResult1.get();
+    Blob remoteBlob2 = getResult2.get();
     assertEquals(remoteUpdatedBlob1, remoteBlob1);
     assertEquals(remoteUpdatedBlob2, remoteBlob2);
 
     // Batch delete request
-    BatchRequest deleteRequest = BatchRequest.builder()
-        .delete(BUCKET, sourceBlobName1)
-        .delete(BUCKET, sourceBlobName2)
-        .build();
-    BatchResponse deleteResponse = storage.submit(deleteRequest);
-    assertEquals(2, deleteResponse.deletes().size());
-    assertEquals(0, deleteResponse.gets().size());
-    assertEquals(0, deleteResponse.updates().size());
-    assertTrue(deleteResponse.deletes().get(0).get());
-    assertTrue(deleteResponse.deletes().get(1).get());
+    StorageBatch deleteBatch = storage.batch();
+    StorageBatchResult<Boolean> deleteResult1 = deleteBatch.delete(BUCKET, sourceBlobName1);
+    StorageBatchResult<Boolean> deleteResult2 = deleteBatch.delete(BUCKET, sourceBlobName2);
+    deleteBatch.submit();
+    assertTrue(deleteResult1.get());
+    assertTrue(deleteResult2.get());
   }
 
   @Test
   public void testBatchRequestManyDeletes() {
     List<BlobId> blobsToDelete = Lists.newArrayListWithCapacity(2 * MAX_BATCH_DELETES);
+    List<StorageBatchResult<Boolean>> deleteResults =
+        Lists.newArrayListWithCapacity(2 * MAX_BATCH_DELETES);
     for (int i = 0; i < 2 * MAX_BATCH_DELETES; i++) {
       blobsToDelete.add(BlobId.of(BUCKET, "test-batch-request-many-deletes-blob-" + i));
     }
-    BatchRequest.Builder builder = BatchRequest.builder();
+    StorageBatch batch = storage.batch();
     for (BlobId blob : blobsToDelete) {
-      builder.delete(blob);
+      deleteResults.add(batch.delete(blob));
     }
     String sourceBlobName1 = "test-batch-request-many-deletes-source-blob-1";
     String sourceBlobName2 = "test-batch-request-many-deletes-source-blob-2";
@@ -840,29 +830,24 @@ public void testBatchRequestManyDeletes() {
     assertNotNull(storage.create(sourceBlob2));
     BlobInfo updatedBlob2 = sourceBlob2.toBuilder().contentType(CONTENT_TYPE).build();
 
-    BatchRequest updateRequest = builder
-        .get(BUCKET, sourceBlobName1)
-        .update(updatedBlob2)
-        .build();
-    BatchResponse response = storage.submit(updateRequest);
-    assertEquals(2 * MAX_BATCH_DELETES, response.deletes().size());
-    assertEquals(1, response.updates().size());
-    assertEquals(1, response.gets().size());
+    StorageBatchResult<Blob> getResult = batch.get(BUCKET, sourceBlobName1);
+    StorageBatchResult<Blob> updateResult = batch.update(updatedBlob2);
+
+    batch.submit();
 
     // Check deletes
-    for (BatchResponse.Result<Boolean> deleteResult : response.deletes()) {
-      assertFalse(deleteResult.failed());
+    for (StorageBatchResult<Boolean> deleteResult : deleteResults) {
       assertFalse(deleteResult.get());
     }
 
     // Check updates
-    Blob remoteUpdatedBlob2 = response.updates().get(0).get();
+    Blob remoteUpdatedBlob2 = updateResult.get();
     assertEquals(sourceBlob2.bucket(), remoteUpdatedBlob2.bucket());
     assertEquals(sourceBlob2.name(), remoteUpdatedBlob2.name());
     assertEquals(updatedBlob2.contentType(), remoteUpdatedBlob2.contentType());
 
     // Check gets
-    Blob remoteBlob1 = response.gets().get(0).get();
+    Blob remoteBlob1 = getResult.get();
     assertEquals(sourceBlob1.bucket(), remoteBlob1.bucket());
     assertEquals(sourceBlob1.name(), remoteBlob1.name());
 
@@ -877,25 +862,36 @@ public void testBatchRequestFail() {
     Blob remoteBlob = storage.create(blob);
     assertNotNull(remoteBlob);
     BlobInfo updatedBlob = BlobInfo.builder(BUCKET, blobName, -1L).build();
-    BatchRequest batchRequest = BatchRequest.builder()
-        .update(updatedBlob, Storage.BlobTargetOption.generationMatch())
-        .delete(BUCKET, blobName, Storage.BlobSourceOption.generationMatch(-1L))
-        .delete(BlobId.of(BUCKET, blobName, -1L))
-        .get(BUCKET, blobName, Storage.BlobGetOption.generationMatch(-1L))
-        .get(BlobId.of(BUCKET, blobName, -1L))
-        .build();
-    BatchResponse batchResponse = storage.submit(batchRequest);
-    assertEquals(1, batchResponse.updates().size());
-    assertEquals(2, batchResponse.deletes().size());
-    assertEquals(2, batchResponse.gets().size());
-    assertTrue(batchResponse.updates().get(0).failed());
-    assertTrue(batchResponse.gets().get(0).failed());
-    assertFalse(batchResponse.gets().get(1).failed());
-    assertNull(batchResponse.gets().get(1).get());
-    assertTrue(batchResponse.deletes().get(0).failed());
-    assertFalse(batchResponse.deletes().get(1).failed());
-    assertFalse(batchResponse.deletes().get(1).get());
-    assertTrue(remoteBlob.delete());
+    StorageBatch batch = storage.batch();
+    StorageBatchResult<Blob> updateResult =
+        batch.update(updatedBlob, Storage.BlobTargetOption.generationMatch());
+    StorageBatchResult<Boolean> deleteResult1 =
+        batch.delete(BUCKET, blobName, Storage.BlobSourceOption.generationMatch(-1L));
+    StorageBatchResult<Boolean> deleteResult2 = batch.delete(BlobId.of(BUCKET, blobName, -1L));
+    StorageBatchResult<Blob> getResult1 =
+        batch.get(BUCKET, blobName, Storage.BlobGetOption.generationMatch(-1L));
+    StorageBatchResult<Blob> getResult2 = batch.get(BlobId.of(BUCKET, blobName, -1L));
+    batch.submit();
+    try {
+      updateResult.get();
+      fail("Expected StorageException");
+    } catch (StorageException ex) {
+      // expected
+    }
+    try {
+      deleteResult1.get();
+      fail("Expected StorageException");
+    } catch (StorageException ex) {
+      // expected
+    }
+    assertFalse(deleteResult2.get());
+    try {
+      getResult1.get();
+      fail("Expected StorageException");
+    } catch (StorageException ex) {
+      // expected
+    }
+    assertNull(getResult2.get());
   }
 
   @Test