From e7054df79b4139fdfd0cc6aa0620fbfa1a10a6b0 Mon Sep 17 00:00:00 2001
From: Brian Chen <chenbrian@google.com>
Date: Tue, 11 Aug 2020 15:42:15 -0500
Subject: [PATCH] feat: add BulkWriter (#323)

New BulkWriter leveraging [`BatchWriteRequest`](https://github.com/googleapis/googleapis/blob/393d03088e368d79c8f4cd5e3610c7df0bb118bf/google/firestore/v1/firestore.proto#L866-L881) which allows for fine grain batch processing of documents. Where as existing `WriteBatch` is an atomic write (all succeed or all fail).

BulkWriter has been implemented to follow the 500/50/5 rule from [Firestore Best Practices: Ramping up traffic](https://firebase.google.com/docs/firestore/best-practices#ramping_up_traffic). If your application is already managing compliance with the 500/50/5 rule, you can configure the `BulkWriter` by using the `Firestore#bulkWriter(com.google.cloud.firestore.BulkWriterOptions)` method.
---
 .../clirr-ignored-differences.xml             |  38 +
 .../cloud/firestore/BatchWriteResult.java     |  55 ++
 .../cloud/firestore/BulkCommitBatch.java      |  63 ++
 .../google/cloud/firestore/BulkWriter.java    | 719 +++++++++++++++++
 .../cloud/firestore/BulkWriterOptions.java    |  43 +
 .../com/google/cloud/firestore/Firestore.java |   6 +
 .../google/cloud/firestore/FirestoreImpl.java |  12 +
 .../google/cloud/firestore/RateLimiter.java   |   4 +-
 .../google/cloud/firestore/Transaction.java   |   4 +
 .../google/cloud/firestore/UpdateBuilder.java | 246 +++++-
 .../google/cloud/firestore/WriteBatch.java    |   4 +
 .../google/cloud/firestore/WriteResult.java   |   2 +-
 .../cloud/firestore/spi/v1/FirestoreRpc.java  |   4 +
 .../firestore/spi/v1/GrpcFirestoreRpc.java    |   7 +
 .../cloud/firestore/BulkWriterTest.java       | 758 ++++++++++++++++++
 .../cloud/firestore/LocalFirestoreHelper.java | 158 +++-
 .../cloud/firestore/TransactionTest.java      |  48 +-
 .../cloud/firestore/WriteBatchTest.java       |  45 +-
 .../cloud/firestore/it/ITSystemTest.java      | 108 +++
 pom.xml                                       |   7 +
 20 files changed, 2254 insertions(+), 77 deletions(-)
 create mode 100644 google-cloud-firestore/src/main/java/com/google/cloud/firestore/BatchWriteResult.java
 create mode 100644 google-cloud-firestore/src/main/java/com/google/cloud/firestore/BulkCommitBatch.java
 create mode 100644 google-cloud-firestore/src/main/java/com/google/cloud/firestore/BulkWriter.java
 create mode 100644 google-cloud-firestore/src/main/java/com/google/cloud/firestore/BulkWriterOptions.java
 create mode 100644 google-cloud-firestore/src/test/java/com/google/cloud/firestore/BulkWriterTest.java

diff --git a/google-cloud-firestore/clirr-ignored-differences.xml b/google-cloud-firestore/clirr-ignored-differences.xml
index 565a807b6..d3fc69ba9 100644
--- a/google-cloud-firestore/clirr-ignored-differences.xml
+++ b/google-cloud-firestore/clirr-ignored-differences.xml
@@ -161,6 +161,44 @@
     <method>com.google.cloud.firestore.Query collectionGroup(java.lang.String)</method>
     <to>com.google.cloud.firestore.CollectionGroup</to>
   </difference>
+  
+  <!--
+  BulkWriter
+  -->
+  <difference>
+    <differenceType>7012</differenceType>
+    <className>com/google/cloud/firestore/Firestore</className>
+    <method>com.google.cloud.firestore.BulkWriter bulkWriter(*)</method>
+  </difference>
+  <difference>
+    <differenceType>7006</differenceType>
+    <className>com/google/cloud/firestore/UpdateBuilder</className>
+    <method>com.google.cloud.firestore.UpdateBuilder create(*)</method>
+    <to>java.lang.Object</to>
+  </difference>
+  <difference>
+    <differenceType>7006</differenceType>
+    <className>com/google/cloud/firestore/UpdateBuilder</className>
+    <method>com.google.cloud.firestore.UpdateBuilder delete(*)</method>
+    <to>java.lang.Object</to>
+  </difference>
+  <difference>
+    <differenceType>7006</differenceType>
+    <className>com/google/cloud/firestore/UpdateBuilder</className>
+    <method>com.google.cloud.firestore.UpdateBuilder set(*)</method>
+    <to>java.lang.Object</to>
+  </difference>
+  <difference>
+    <differenceType>7006</differenceType>
+    <className>com/google/cloud/firestore/UpdateBuilder</className>
+    <method>com.google.cloud.firestore.UpdateBuilder update(*)</method>
+    <to>java.lang.Object</to>
+  </difference>
+  <difference>
+    <differenceType>7012</differenceType>
+    <className>com/google/cloud/firestore/spi/v1/FirestoreRpc</className>
+    <method>com.google.api.gax.rpc.UnaryCallable batchWriteCallable()</method>
+  </difference>
 
   <!--
   FakeCredentials Refactor
diff --git a/google-cloud-firestore/src/main/java/com/google/cloud/firestore/BatchWriteResult.java b/google-cloud-firestore/src/main/java/com/google/cloud/firestore/BatchWriteResult.java
new file mode 100644
index 000000000..96ce5d60b
--- /dev/null
+++ b/google-cloud-firestore/src/main/java/com/google/cloud/firestore/BatchWriteResult.java
@@ -0,0 +1,55 @@
+/*
+ * Copyright 2020 Google LLC
+ *
+ * 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.firestore;
+
+import com.google.api.core.InternalApi;
+import com.google.cloud.Timestamp;
+import javax.annotation.Nullable;
+
+/**
+ * A BatchWriteResult wraps the write time and status returned by Firestore when making
+ * BatchWriteRequests.
+ */
+@InternalApi
+public final class BatchWriteResult {
+  private final DocumentReference documentReference;
+  @Nullable private final Timestamp writeTime;
+  @Nullable private final Exception exception;
+
+  BatchWriteResult(
+      DocumentReference documentReference,
+      @Nullable Timestamp timestamp,
+      @Nullable Exception exception) {
+    this.documentReference = documentReference;
+    this.writeTime = timestamp;
+    this.exception = exception;
+  }
+
+  public DocumentReference getDocumentReference() {
+    return documentReference;
+  }
+
+  @Nullable
+  public Timestamp getWriteTime() {
+    return writeTime;
+  }
+
+  @Nullable
+  public Exception getException() {
+    return exception;
+  }
+}
diff --git a/google-cloud-firestore/src/main/java/com/google/cloud/firestore/BulkCommitBatch.java b/google-cloud-firestore/src/main/java/com/google/cloud/firestore/BulkCommitBatch.java
new file mode 100644
index 000000000..701e6114f
--- /dev/null
+++ b/google-cloud-firestore/src/main/java/com/google/cloud/firestore/BulkCommitBatch.java
@@ -0,0 +1,63 @@
+/*
+ * Copyright 2020 Google LLC
+ *
+ * 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.firestore;
+
+import com.google.api.core.ApiFuture;
+import com.google.common.base.Preconditions;
+import com.google.common.base.Predicate;
+import com.google.common.collect.FluentIterable;
+import java.util.Set;
+
+/** Used to represent a batch on the BatchQueue. */
+class BulkCommitBatch extends UpdateBuilder<ApiFuture<WriteResult>> {
+
+  BulkCommitBatch(FirestoreImpl firestore, int maxBatchSize) {
+    super(firestore, maxBatchSize);
+  }
+
+  BulkCommitBatch(
+      FirestoreImpl firestore,
+      BulkCommitBatch retryBatch,
+      final Set<DocumentReference> docsToRetry) {
+    super(firestore);
+    this.writes.addAll(
+        FluentIterable.from(retryBatch.writes)
+            .filter(
+                new Predicate<WriteOperation>() {
+                  @Override
+                  public boolean apply(WriteOperation writeOperation) {
+                    return docsToRetry.contains(writeOperation.documentReference);
+                  }
+                })
+            .toList());
+
+    Preconditions.checkState(
+        retryBatch.state == BatchState.SENT,
+        "Batch should be SENT when creating a new BulkCommitBatch for retry");
+    this.state = retryBatch.state;
+    this.pendingOperations = retryBatch.pendingOperations;
+  }
+
+  ApiFuture<WriteResult> wrapResult(ApiFuture<WriteResult> result) {
+    return result;
+  }
+
+  @Override
+  boolean allowDuplicateDocs() {
+    return false;
+  }
+}
diff --git a/google-cloud-firestore/src/main/java/com/google/cloud/firestore/BulkWriter.java b/google-cloud-firestore/src/main/java/com/google/cloud/firestore/BulkWriter.java
new file mode 100644
index 000000000..6df7d0d2e
--- /dev/null
+++ b/google-cloud-firestore/src/main/java/com/google/cloud/firestore/BulkWriter.java
@@ -0,0 +1,719 @@
+/*
+ * Copyright 2020 Google LLC
+ *
+ * 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.firestore;
+
+import com.google.api.core.ApiAsyncFunction;
+import com.google.api.core.ApiFuture;
+import com.google.api.core.ApiFutures;
+import com.google.api.core.CurrentMillisClock;
+import com.google.api.core.SettableApiFuture;
+import com.google.api.gax.retrying.ExponentialRetryAlgorithm;
+import com.google.api.gax.retrying.TimedAttemptSettings;
+import com.google.cloud.firestore.UpdateBuilder.BatchState;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import com.google.common.base.Predicate;
+import com.google.common.collect.FluentIterable;
+import com.google.common.util.concurrent.MoreExecutors;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+public final class BulkWriter implements AutoCloseable {
+  /** The maximum number of writes that can be in a single batch. */
+  public static final int MAX_BATCH_SIZE = 500;
+
+  public static final int MAX_RETRY_ATTEMPTS = 10;
+
+  /**
+   * The starting maximum number of operations per second as allowed by the 500/50/5 rule.
+   *
+   * @see <a href=https://cloud.google.com/datastore/docs/best-practices#ramping_up_traffic>Ramping
+   *     up traffic</a>
+   */
+  private static final int STARTING_MAXIMUM_OPS_PER_SECOND = 500;
+
+  /**
+   * The rate by which to increase the capacity as specified by the 500/50/5 rule.
+   *
+   * @see <a href=https://cloud.google.com/datastore/docs/best-practices#ramping_up_traffic>Ramping
+   *     up traffic</a>
+   */
+  private static final double RATE_LIMITER_MULTIPLIER = 1.5;
+
+  /**
+   * How often the operations per second capacity should increase in milliseconds as specified by
+   * the 500/50/5 rule.
+   *
+   * @see <a href=https://cloud.google.com/datastore/docs/best-practices#ramping_up_traffic>Ramping
+   *     up traffic</a>
+   */
+  private static final int RATE_LIMITER_MULTIPLIER_MILLIS = 5 * 60 * 1000;
+
+  private static final Logger logger = Logger.getLogger(BulkWriter.class.getName());
+
+  /** The maximum number of writes that can be in a single batch. */
+  private int maxBatchSize = MAX_BATCH_SIZE;
+
+  /**
+   * A queue of batches to be written. Use a synchronized list to avoid multi-thread concurrent
+   * modification errors (as this list is modified from both the user thread and the network
+   * thread).
+   */
+  private final List<BulkCommitBatch> batchQueue = new CopyOnWriteArrayList<>();
+
+  /** Whether this BulkWriter instance is closed. Once closed, it cannot be opened again. */
+  private boolean closed = false;
+
+  /** Rate limiter used to throttle requests as per the 500/50/5 rule. */
+  private final RateLimiter rateLimiter;
+
+  private final FirestoreImpl firestore;
+
+  private final ScheduledExecutorService firestoreExecutor;
+
+  private final ExponentialRetryAlgorithm backoff;
+  private TimedAttemptSettings nextAttempt;
+
+  BulkWriter(FirestoreImpl firestore, boolean enableThrottling) {
+    this.firestore = firestore;
+    this.backoff =
+        new ExponentialRetryAlgorithm(
+            firestore.getOptions().getRetrySettings(), CurrentMillisClock.getDefaultClock());
+    this.nextAttempt = backoff.createFirstAttempt();
+    this.firestoreExecutor = firestore.getClient().getExecutor();
+
+    if (enableThrottling) {
+      rateLimiter =
+          new RateLimiter(
+              STARTING_MAXIMUM_OPS_PER_SECOND,
+              RATE_LIMITER_MULTIPLIER,
+              RATE_LIMITER_MULTIPLIER_MILLIS);
+    } else {
+      rateLimiter = new RateLimiter(Integer.MAX_VALUE, Double.MAX_VALUE, Integer.MAX_VALUE);
+    }
+  }
+
+  /**
+   * Create a document with the provided data. This single operation will fail if a document exists
+   * at its location.
+   *
+   * @param documentReference A reference to the document to be created.
+   * @param fields A map of the fields and values for the document.
+   * @return An ApiFuture containing the result of the write. Contains an error if the write fails.
+   */
+  @Nonnull
+  public ApiFuture<WriteResult> create(
+      @Nonnull DocumentReference documentReference, @Nonnull Map<String, Object> fields) {
+    verifyNotClosed();
+    BulkCommitBatch bulkCommitBatch = getEligibleBatch(documentReference);
+    ApiFuture<WriteResult> future = bulkCommitBatch.create(documentReference, fields);
+    sendReadyBatches();
+    return future;
+  }
+
+  /**
+   * Create a document with the provided data. This single operation will fail if a document exists
+   * at its location.
+   *
+   * @param documentReference A reference to the document to be created.
+   * @param pojo The POJO that will be used to populate the document contents.
+   * @return An ApiFuture containing the result of the write. Contains an error if the write fails.
+   */
+  @Nonnull
+  public ApiFuture<WriteResult> create(
+      @Nonnull DocumentReference documentReference, @Nonnull Object pojo) {
+    verifyNotClosed();
+    BulkCommitBatch bulkCommitBatch = getEligibleBatch(documentReference);
+    ApiFuture<WriteResult> future = bulkCommitBatch.create(documentReference, pojo);
+    sendReadyBatches();
+    return future;
+  }
+
+  /**
+   * Delete a document from the database.
+   *
+   * @param documentReference The DocumentReference to delete.
+   * @return An ApiFuture containing the result of the delete. Contains an error if the delete
+   *     fails.
+   */
+  @Nonnull
+  public ApiFuture<WriteResult> delete(@Nonnull DocumentReference documentReference) {
+    verifyNotClosed();
+    BulkCommitBatch bulkCommitBatch = getEligibleBatch(documentReference);
+    ApiFuture<WriteResult> future = bulkCommitBatch.delete(documentReference);
+    sendReadyBatches();
+    return future;
+  }
+
+  /**
+   * Delete a document from the database.
+   *
+   * @param documentReference The DocumentReference to delete.
+   * @param precondition Precondition to enforce for this delete.
+   * @return An ApiFuture containing a sentinel value (Timestamp(0)) for the delete operation.
+   *     Contains an error if the delete fails.
+   */
+  @Nonnull
+  public ApiFuture<WriteResult> delete(
+      @Nonnull DocumentReference documentReference, @Nonnull Precondition precondition) {
+    verifyNotClosed();
+    BulkCommitBatch bulkCommitBatch = getEligibleBatch(documentReference);
+    ApiFuture<WriteResult> future = bulkCommitBatch.delete(documentReference, precondition);
+    sendReadyBatches();
+    return future;
+  }
+
+  /**
+   * Write to the document referred to by the provided DocumentReference. If the document does not
+   * exist yet, it will be created.
+   *
+   * @param documentReference A reference to the document to be set.
+   * @param fields A map of the fields and values for the document.
+   * @return An ApiFuture containing a sentinel value (Timestamp(0)) for the delete operation.
+   *     Contains an error if the delete fails.
+   */
+  @Nonnull
+  public ApiFuture<WriteResult> set(
+      @Nonnull DocumentReference documentReference, @Nonnull Map<String, Object> fields) {
+    verifyNotClosed();
+    BulkCommitBatch bulkCommitBatch = getEligibleBatch(documentReference);
+    ApiFuture<WriteResult> future = bulkCommitBatch.set(documentReference, fields);
+    sendReadyBatches();
+    return future;
+  }
+
+  /**
+   * Write to the document referred to by the provided DocumentReference. If the document does not
+   * exist yet, it will be created. If you pass a {@link SetOptions}, the provided data can be
+   * merged into an existing document.
+   *
+   * @param documentReference A reference to the document to be set.
+   * @param fields A map of the fields and values for the document.
+   * @param options An object to configure the set behavior.
+   * @return An ApiFuture containing the result of the write. Contains an error if the write fails.
+   */
+  @Nonnull
+  public ApiFuture<WriteResult> set(
+      @Nonnull DocumentReference documentReference,
+      @Nonnull Map<String, Object> fields,
+      @Nonnull SetOptions options) {
+    verifyNotClosed();
+    BulkCommitBatch bulkCommitBatch = getEligibleBatch(documentReference);
+    ApiFuture<WriteResult> future = bulkCommitBatch.set(documentReference, fields, options);
+    sendReadyBatches();
+    return future;
+  }
+
+  /**
+   * Write to the document referred to by the provided DocumentReference. If the document does not
+   * exist yet, it will be created. If you pass a {@link SetOptions}, the provided data can be
+   * merged into an existing document.
+   *
+   * @param documentReference A reference to the document to be set.
+   * @param pojo The POJO that will be used to populate the document contents.
+   * @param options An object to configure the set behavior.
+   * @return An ApiFuture containing the result of the write. Contains an error if the write fails.
+   */
+  @Nonnull
+  public ApiFuture<WriteResult> set(
+      @Nonnull DocumentReference documentReference, Object pojo, @Nonnull SetOptions options) {
+    verifyNotClosed();
+    BulkCommitBatch bulkCommitBatch = getEligibleBatch(documentReference);
+    ApiFuture<WriteResult> future = bulkCommitBatch.set(documentReference, pojo, options);
+    sendReadyBatches();
+    return future;
+  }
+
+  /**
+   * Write to the document referred to by the provided DocumentReference. If the document does not
+   * exist yet, it will be created.
+   *
+   * @param documentReference A reference to the document to be set.
+   * @param pojo The POJO that will be used to populate the document contents.
+   * @return An ApiFuture containing the result of the write. Contains an error if the write fails.
+   */
+  @Nonnull
+  public ApiFuture<WriteResult> set(@Nonnull DocumentReference documentReference, Object pojo) {
+    verifyNotClosed();
+    BulkCommitBatch bulkCommitBatch = getEligibleBatch(documentReference);
+    ApiFuture<WriteResult> future = bulkCommitBatch.set(documentReference, pojo);
+    sendReadyBatches();
+    return future;
+  }
+
+  /**
+   * Update fields of the document referred to by the provided {@link DocumentReference}. If the
+   * document doesn't yet exist, the update will fail.
+   *
+   * <p>The update() method accepts either an object with field paths encoded as keys and field
+   * values encoded as values, or a variable number of arguments that alternate between field paths
+   * and field values. Nested fields can be updated by providing dot-separated field path strings or
+   * by providing FieldPath objects.
+   *
+   * @param documentReference A reference to the document to be updated.
+   * @param fields A map of the fields and values for the document.
+   * @return An ApiFuture containing the result of the write. Contains an error if the write fails.
+   */
+  @Nonnull
+  public ApiFuture<WriteResult> update(
+      @Nonnull DocumentReference documentReference, @Nonnull Map<String, Object> fields) {
+    verifyNotClosed();
+    BulkCommitBatch bulkCommitBatch = getEligibleBatch(documentReference);
+    ApiFuture<WriteResult> future = bulkCommitBatch.update(documentReference, fields);
+    sendReadyBatches();
+    return future;
+  }
+
+  /**
+   * Update fields of the document referred to by the provided {@link DocumentReference}. If the
+   * document doesn't yet exist, the update will fail.
+   *
+   * <p>The update() method accepts either an object with field paths encoded as keys and field
+   * values encoded as values, or a variable number of arguments that alternate between field paths
+   * and field values. Nested fields can be updated by providing dot-separated field path strings or
+   * by providing FieldPath objects.
+   *
+   * @param documentReference A reference to the document to be updated.
+   * @param fields A map of the fields and values for the document.
+   * @param precondition Precondition to enforce on this update.
+   * @return An ApiFuture containing the result of the write. Contains an error if the write fails.
+   */
+  @Nonnull
+  public ApiFuture<WriteResult> update(
+      @Nonnull DocumentReference documentReference,
+      @Nonnull Map<String, Object> fields,
+      Precondition precondition) {
+    verifyNotClosed();
+    BulkCommitBatch bulkCommitBatch = getEligibleBatch(documentReference);
+    ApiFuture<WriteResult> future = bulkCommitBatch.update(documentReference, fields, precondition);
+    sendReadyBatches();
+    return future;
+  }
+
+  /**
+   * Update fields of the document referred to by the provided {@link DocumentReference}. If the
+   * document doesn't yet exist, the update will fail.
+   *
+   * <p>The update() method accepts either an object with field paths encoded as keys and field
+   * values encoded as values, or a variable number of arguments that alternate between field paths
+   * and field values. Nested fields can be updated by providing dot-separated field path strings or
+   * by providing FieldPath objects.
+   *
+   * @param documentReference A reference to the document to be updated.
+   * @param field The first field to set.
+   * @param value The first value to set.
+   * @param moreFieldsAndValues String and Object pairs with more fields to be set.
+   * @return An ApiFuture containing the result of the write. Contains an error if the write fails.
+   */
+  @Nonnull
+  public ApiFuture<WriteResult> update(
+      @Nonnull DocumentReference documentReference,
+      @Nonnull String field,
+      @Nullable Object value,
+      Object... moreFieldsAndValues) {
+    verifyNotClosed();
+    BulkCommitBatch bulkCommitBatch = getEligibleBatch(documentReference);
+    ApiFuture<WriteResult> future =
+        bulkCommitBatch.update(documentReference, field, value, moreFieldsAndValues);
+    sendReadyBatches();
+    return future;
+  }
+
+  /**
+   * Update fields of the document referred to by the provided {@link DocumentReference}. If the
+   * document doesn't yet exist, the update will fail.
+   *
+   * <p>The update() method accepts either an object with field paths encoded as keys and field
+   * values encoded as values, or a variable number of arguments that alternate between field paths
+   * and field values. Nested fields can be updated by providing dot-separated field path strings or
+   * by providing FieldPath objects.
+   *
+   * @param documentReference A reference to the document to be updated.
+   * @param fieldPath The first field to set.
+   * @param value The first value to set.
+   * @param moreFieldsAndValues String and Object pairs with more fields to be set.
+   * @return An ApiFuture containing the result of the write. Contains an error if the write fails.
+   */
+  @Nonnull
+  public ApiFuture<WriteResult> update(
+      @Nonnull DocumentReference documentReference,
+      @Nonnull FieldPath fieldPath,
+      @Nullable Object value,
+      Object... moreFieldsAndValues) {
+    verifyNotClosed();
+    BulkCommitBatch bulkCommitBatch = getEligibleBatch(documentReference);
+    ApiFuture<WriteResult> future =
+        bulkCommitBatch.update(documentReference, fieldPath, value, moreFieldsAndValues);
+    sendReadyBatches();
+    return future;
+  }
+
+  /**
+   * Update fields of the document referred to by the provided {@link DocumentReference}. If the
+   * document doesn't yet exist, the update will fail.
+   *
+   * <p>The update() method accepts either an object with field paths encoded as keys and field
+   * values encoded as values, or a variable number of arguments that alternate between field paths
+   * and field values. Nested fields can be updated by providing dot-separated field path strings or
+   * by providing FieldPath objects.
+   *
+   * @param documentReference A reference to the document to be updated.
+   * @param field The first field to set.
+   * @param value The first value to set.
+   * @param moreFieldsAndValues String and Object pairs with more fields to be set.
+   * @return An ApiFuture containing the result of the write. Contains an error if the write fails.
+   */
+  @Nonnull
+  public ApiFuture<WriteResult> update(
+      @Nonnull DocumentReference documentReference,
+      @Nonnull Precondition precondition,
+      @Nonnull String field,
+      @Nullable Object value,
+      Object... moreFieldsAndValues) {
+    verifyNotClosed();
+    BulkCommitBatch bulkCommitBatch = getEligibleBatch(documentReference);
+    ApiFuture<WriteResult> future =
+        bulkCommitBatch.update(documentReference, precondition, field, value, moreFieldsAndValues);
+    sendReadyBatches();
+    return future;
+  }
+
+  /**
+   * Update fields of the document referred to by the provided {@link DocumentReference}. If the
+   * document doesn't yet exist, the update will fail.
+   *
+   * <p>The update() method accepts either an object with field paths encoded as keys and field
+   * values encoded as values, or a variable number of arguments that alternate between field paths
+   * and field values. Nested fields can be updated by providing dot-separated field path strings or
+   * by providing FieldPath objects.
+   *
+   * @param documentReference A reference to the document to be updated.
+   * @param precondition Precondition to enforce on this update.
+   * @param fieldPath The first field to set.
+   * @param value The first value to set.
+   * @param moreFieldsAndValues String and Object pairs with more fields to be set.
+   * @return An ApiFuture containing the result of the write. Contains an error if the write fails.
+   */
+  @Nonnull
+  public ApiFuture<WriteResult> update(
+      @Nonnull DocumentReference documentReference,
+      @Nonnull Precondition precondition,
+      @Nonnull FieldPath fieldPath,
+      @Nullable Object value,
+      Object... moreFieldsAndValues) {
+    verifyNotClosed();
+    BulkCommitBatch bulkCommitBatch = getEligibleBatch(documentReference);
+    ApiFuture<WriteResult> future =
+        bulkCommitBatch.update(
+            documentReference, precondition, fieldPath, value, moreFieldsAndValues);
+    sendReadyBatches();
+    return future;
+  }
+
+  /**
+   * Commits all writes that have been enqueued up to this point in parallel.
+   *
+   * <p>Returns an ApiFuture that completes when all currently queued operations have been
+   * committed. The ApiFuture will never return an error since the results for each individual
+   * operation are conveyed via their individual ApiFutures.
+   *
+   * <p>The ApiFuture completes immediately if there are no pending writes. Otherwise, the ApiFuture
+   * waits for all previously issued writes, but it does not wait for writes that were added after
+   * the method is called. If you want to wait for additional writes, call `flush()` again.
+   *
+   * @return An ApiFuture that completes when all enqueued writes up to this point have been
+   *     committed.
+   */
+  @Nonnull
+  public ApiFuture<Void> flush() {
+    verifyNotClosed();
+    final SettableApiFuture<Void> flushComplete = SettableApiFuture.create();
+    List<SettableApiFuture<WriteResult>> writeFutures = new ArrayList<>();
+    for (BulkCommitBatch batch : batchQueue) {
+      batch.markReadyToSend();
+      writeFutures.addAll(batch.getPendingFutures());
+    }
+    sendReadyBatches();
+    ApiFutures.successfulAsList(writeFutures)
+        .addListener(
+            new Runnable() {
+              public void run() {
+                flushComplete.set(null);
+              }
+            },
+            MoreExecutors.directExecutor());
+    return flushComplete;
+  }
+
+  /**
+   * Commits all enqueued writes and marks the BulkWriter instance as closed.
+   *
+   * <p>After calling `close()`, calling any method wil return an error.
+   *
+   * <p>This method completes when there are no more pending writes. Calling this method will send
+   * all requests.
+   */
+  public void close() throws InterruptedException, ExecutionException {
+    ApiFuture<Void> flushFuture = flush();
+    closed = true;
+    flushFuture.get();
+  }
+
+  private void verifyNotClosed() {
+    if (this.closed) {
+      throw new IllegalStateException("BulkWriter has already been closed.");
+    }
+  }
+
+  /**
+   * Return the first eligible batch that can hold a write to the provided reference, or creates one
+   * if no eligible batches are found.
+   */
+  private BulkCommitBatch getEligibleBatch(DocumentReference documentReference) {
+    if (batchQueue.size() > 0) {
+      BulkCommitBatch lastBatch = batchQueue.get(batchQueue.size() - 1);
+      if (lastBatch.getState() == UpdateBuilder.BatchState.OPEN
+          && !lastBatch.hasDocument(documentReference)) {
+        return lastBatch;
+      }
+    }
+    return createNewBatch();
+  }
+
+  /**
+   * Creates a new batch and adds it to the BatchQueue. If there is already a batch enqueued, sends
+   * the batch after a new one is created.
+   */
+  private BulkCommitBatch createNewBatch() {
+    BulkCommitBatch newBatch = new BulkCommitBatch(firestore, maxBatchSize);
+
+    if (batchQueue.size() > 0) {
+      batchQueue.get(batchQueue.size() - 1).markReadyToSend();
+      sendReadyBatches();
+    }
+    batchQueue.add(newBatch);
+    return newBatch;
+  }
+
+  /**
+   * Attempts to send batches starting from the front of the BatchQueue until a batch cannot be
+   * sent.
+   *
+   * <p>After a batch is complete, try sending batches again.
+   */
+  private void sendReadyBatches() {
+    List<BulkCommitBatch> unsentBatches =
+        FluentIterable.from(batchQueue)
+            .filter(
+                new Predicate<BulkCommitBatch>() {
+                  @Override
+                  public boolean apply(BulkCommitBatch batch) {
+                    return batch.getState() == UpdateBuilder.BatchState.READY_TO_SEND;
+                  }
+                })
+            .toList();
+
+    int index = 0;
+    while (index < unsentBatches.size() && isBatchSendable(unsentBatches.get(index))) {
+      final BulkCommitBatch batch = unsentBatches.get(index);
+
+      // Send the batch if it is under the rate limit, or schedule another attempt after the
+      // appropriate timeout.
+      long delayMs = rateLimiter.getNextRequestDelayMs(batch.getPendingOperationCount());
+      Preconditions.checkState(delayMs != -1, "Batch size should be under capacity");
+      if (delayMs == 0) {
+        sendBatch(batch);
+      } else {
+        firestoreExecutor.schedule(
+            new Runnable() {
+              @Override
+              public void run() {
+                sendBatch(batch);
+              }
+            },
+            delayMs,
+            TimeUnit.MILLISECONDS);
+        break;
+      }
+
+      ++index;
+    }
+  }
+
+  /**
+   * Sends the provided batch and processes the results. After the batch is committed, sends the
+   * next group of ready batches.
+   */
+  private void sendBatch(final BulkCommitBatch batch) {
+    Preconditions.checkState(
+        batch.state == BatchState.READY_TO_SEND,
+        "The batch should be marked as READY_TO_SEND before committing");
+    batch.state = BatchState.SENT;
+    boolean success = rateLimiter.tryMakeRequest(batch.getPendingOperationCount());
+    Preconditions.checkState(success, "Batch should be under rate limit to be sent.");
+
+    ApiFuture<Void> commitFuture = bulkCommit(batch);
+    commitFuture.addListener(
+        new Runnable() {
+          public void run() {
+            boolean removed = batchQueue.remove(batch);
+            Preconditions.checkState(
+                removed, "The batch should be in the BatchQueue." + batchQueue.size());
+            sendReadyBatches();
+          }
+        },
+        MoreExecutors.directExecutor());
+  }
+
+  private ApiFuture<Void> bulkCommit(BulkCommitBatch batch) {
+    return bulkCommit(batch, 0);
+  }
+
+  private ApiFuture<Void> bulkCommit(final BulkCommitBatch batch, final int attempt) {
+    final SettableApiFuture<Void> backoffFuture = SettableApiFuture.create();
+
+    // Add a backoff delay. At first, this is 0.
+    firestoreExecutor.schedule(
+        new Runnable() {
+          @Override
+          public void run() {
+            backoffFuture.set(null);
+          }
+        },
+        nextAttempt.getRandomizedRetryDelay().toMillis(),
+        TimeUnit.MILLISECONDS);
+
+    return ApiFutures.transformAsync(
+        backoffFuture, new BackoffCallback(batch, attempt), firestoreExecutor);
+  }
+
+  private class BackoffCallback implements ApiAsyncFunction<Void, Void> {
+    final BulkCommitBatch batch;
+    final int attempt;
+
+    public BackoffCallback(BulkCommitBatch batch, int attempt) {
+      this.batch = batch;
+      this.attempt = attempt;
+    }
+
+    @Override
+    public ApiFuture<Void> apply(Void ignored) {
+
+      return ApiFutures.transformAsync(
+          ApiFutures.catchingAsync(
+              batch.bulkCommit(),
+              Exception.class,
+              new ApiAsyncFunction<Exception, List<BatchWriteResult>>() {
+                public ApiFuture<List<BatchWriteResult>> apply(Exception exception) {
+                  List<BatchWriteResult> results = new ArrayList<>();
+                  // If the BatchWrite RPC fails, map the exception to each individual result.
+                  for (DocumentReference documentReference : batch.getPendingDocuments()) {
+                    results.add(new BatchWriteResult(documentReference, null, exception));
+                  }
+                  return ApiFutures.immediateFuture(results);
+                }
+              },
+              MoreExecutors.directExecutor()),
+          new ProcessBulkCommitCallback(batch, attempt),
+          MoreExecutors.directExecutor());
+    }
+  }
+
+  private class ProcessBulkCommitCallback
+      implements ApiAsyncFunction<List<BatchWriteResult>, Void> {
+    final BulkCommitBatch batch;
+    final int attempt;
+
+    public ProcessBulkCommitCallback(BulkCommitBatch batch, int attempt) {
+      this.batch = batch;
+      this.attempt = attempt;
+    }
+
+    @Override
+    public ApiFuture<Void> apply(List<BatchWriteResult> results) {
+      batch.processResults(results);
+      Set<DocumentReference> remainingOps = batch.getPendingDocuments();
+      if (!remainingOps.isEmpty()) {
+        logger.log(
+            Level.WARNING,
+            String.format(
+                "Current batch failed at retry #%d. Num failures: %d",
+                attempt, remainingOps.size()));
+
+        if (attempt < MAX_RETRY_ATTEMPTS) {
+          nextAttempt = backoff.createNextAttempt(nextAttempt);
+          BulkCommitBatch newBatch = new BulkCommitBatch(firestore, batch, remainingOps);
+          return bulkCommit(newBatch, attempt + 1);
+        } else {
+          batch.failRemainingOperations(results);
+        }
+      }
+      return ApiFutures.immediateFuture(null);
+    }
+  }
+
+  /**
+   * Checks that the provided batch is sendable. To be sendable, a batch must: (1) be marked as
+   * READY_TO_SEND (2) not write to references that are currently in flight.
+   */
+  private boolean isBatchSendable(BulkCommitBatch batch) {
+    if (!batch.getState().equals(UpdateBuilder.BatchState.READY_TO_SEND)) {
+      return false;
+    }
+
+    for (final DocumentReference documentReference : batch.getPendingDocuments()) {
+      boolean isRefInFlight =
+          FluentIterable.from(batchQueue)
+              .anyMatch(
+                  new Predicate<BulkCommitBatch>() {
+                    @Override
+                    public boolean apply(BulkCommitBatch batch) {
+                      return batch.getState().equals(BatchState.SENT)
+                          && batch.hasDocument(documentReference);
+                    }
+                  });
+
+      if (isRefInFlight) {
+        logger.log(
+            Level.WARNING,
+            String.format(
+                "Duplicate write to document %s detected. Writing to the same document multiple"
+                    + " times will slow down BulkWriter. Write to unique documents in order to "
+                    + "maximize throughput.",
+                documentReference.getPath()));
+        return false;
+      }
+    }
+
+    return true;
+  }
+
+  @VisibleForTesting
+  void setMaxBatchSize(int size) {
+    maxBatchSize = size;
+  }
+}
diff --git a/google-cloud-firestore/src/main/java/com/google/cloud/firestore/BulkWriterOptions.java b/google-cloud-firestore/src/main/java/com/google/cloud/firestore/BulkWriterOptions.java
new file mode 100644
index 000000000..26f57f086
--- /dev/null
+++ b/google-cloud-firestore/src/main/java/com/google/cloud/firestore/BulkWriterOptions.java
@@ -0,0 +1,43 @@
+/*
+ * Copyright 2020 Google LLC
+ *
+ * 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.firestore;
+
+import javax.annotation.Nonnull;
+
+/** Options used to disable request throttling in BulkWriter. */
+public final class BulkWriterOptions {
+
+  private final boolean enableThrottling;
+
+  private BulkWriterOptions(boolean enableThrottling) {
+    this.enableThrottling = enableThrottling;
+  }
+
+  boolean isThrottlingEnabled() {
+    return enableThrottling;
+  }
+
+  /**
+   * An options object that will disable throttling in the created BulkWriter.
+   *
+   * @return The BulkWriterOptions object.
+   */
+  @Nonnull
+  public static BulkWriterOptions withThrottlingDisabled() {
+    return new BulkWriterOptions(false);
+  }
+}
diff --git a/google-cloud-firestore/src/main/java/com/google/cloud/firestore/Firestore.java b/google-cloud-firestore/src/main/java/com/google/cloud/firestore/Firestore.java
index e69a2f903..aab60b091 100644
--- a/google-cloud-firestore/src/main/java/com/google/cloud/firestore/Firestore.java
+++ b/google-cloud-firestore/src/main/java/com/google/cloud/firestore/Firestore.java
@@ -168,6 +168,12 @@ void getAll(
   @Nonnull
   WriteBatch batch();
 
+  @Nonnull
+  BulkWriter bulkWriter();
+
+  @Nonnull
+  BulkWriter bulkWriter(BulkWriterOptions options);
+
   /**
    * Closes the gRPC channels associated with this instance and frees up their resources. This
    * method blocks until all channels are closed. Once this method is called, this Firestore client
diff --git a/google-cloud-firestore/src/main/java/com/google/cloud/firestore/FirestoreImpl.java b/google-cloud-firestore/src/main/java/com/google/cloud/firestore/FirestoreImpl.java
index b33fa2747..717cf8fe4 100644
--- a/google-cloud-firestore/src/main/java/com/google/cloud/firestore/FirestoreImpl.java
+++ b/google-cloud-firestore/src/main/java/com/google/cloud/firestore/FirestoreImpl.java
@@ -94,6 +94,18 @@ public WriteBatch batch() {
     return new WriteBatch(this);
   }
 
+  @Nonnull
+  @Override
+  public BulkWriter bulkWriter() {
+    return new BulkWriter(this, /* enableThrottling= */ true);
+  }
+
+  @Nonnull
+  @Override
+  public BulkWriter bulkWriter(BulkWriterOptions options) {
+    return new BulkWriter(this, options.isThrottlingEnabled());
+  }
+
   @Nonnull
   @Override
   public CollectionReference collection(@Nonnull String collectionPath) {
diff --git a/google-cloud-firestore/src/main/java/com/google/cloud/firestore/RateLimiter.java b/google-cloud-firestore/src/main/java/com/google/cloud/firestore/RateLimiter.java
index 7e617c546..8bd6e5ce3 100644
--- a/google-cloud-firestore/src/main/java/com/google/cloud/firestore/RateLimiter.java
+++ b/google-cloud-firestore/src/main/java/com/google/cloud/firestore/RateLimiter.java
@@ -42,7 +42,7 @@ class RateLimiter {
   private int availableTokens;
   private long lastRefillTimeMillis;
 
-  RateLimiter(int initialCapacity, int multiplier, int multiplierMillis) {
+  RateLimiter(int initialCapacity, double multiplier, int multiplierMillis) {
     this(initialCapacity, multiplier, multiplierMillis, new Date().getTime());
   }
 
@@ -64,7 +64,7 @@ class RateLimiter {
   }
 
   public boolean tryMakeRequest(int numOperations) {
-    return tryMakeRequest(numOperations, new Date(0).getTime());
+    return tryMakeRequest(numOperations, new Date().getTime());
   }
 
   /**
diff --git a/google-cloud-firestore/src/main/java/com/google/cloud/firestore/Transaction.java b/google-cloud-firestore/src/main/java/com/google/cloud/firestore/Transaction.java
index 27b0ccd0c..c543c3743 100644
--- a/google-cloud-firestore/src/main/java/com/google/cloud/firestore/Transaction.java
+++ b/google-cloud-firestore/src/main/java/com/google/cloud/firestore/Transaction.java
@@ -69,6 +69,10 @@ public interface AsyncFunction<T> {
     previousTransactionId = previousTransaction != null ? previousTransaction.transactionId : null;
   }
 
+  Transaction wrapResult(ApiFuture<WriteResult> result) {
+    return this;
+  }
+
   /** Starts a transaction and obtains the transaction id. */
   ApiFuture<Void> begin() {
     BeginTransactionRequest.Builder beginTransaction = BeginTransactionRequest.newBuilder();
diff --git a/google-cloud-firestore/src/main/java/com/google/cloud/firestore/UpdateBuilder.java b/google-cloud-firestore/src/main/java/com/google/cloud/firestore/UpdateBuilder.java
index c80916099..7439fe081 100644
--- a/google-cloud-firestore/src/main/java/com/google/cloud/firestore/UpdateBuilder.java
+++ b/google-cloud-firestore/src/main/java/com/google/cloud/firestore/UpdateBuilder.java
@@ -19,21 +19,31 @@
 import com.google.api.core.ApiFunction;
 import com.google.api.core.ApiFuture;
 import com.google.api.core.ApiFutures;
+import com.google.api.core.InternalExtensionOnly;
+import com.google.api.core.SettableApiFuture;
+import com.google.api.gax.rpc.StatusCode.Code;
+import com.google.cloud.Timestamp;
 import com.google.cloud.firestore.UserDataConverter.EncodingOptions;
+import com.google.cloud.firestore.v1.FirestoreSettings;
 import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.util.concurrent.MoreExecutors;
+import com.google.firestore.v1.BatchWriteRequest;
+import com.google.firestore.v1.BatchWriteResponse;
 import com.google.firestore.v1.CommitRequest;
 import com.google.firestore.v1.CommitResponse;
 import com.google.firestore.v1.Write;
 import com.google.protobuf.ByteString;
+import io.grpc.Status;
 import io.opencensus.trace.AttributeValue;
 import io.opencensus.trace.Tracing;
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
+import java.util.Set;
 import java.util.SortedSet;
 import java.util.TreeSet;
 import javax.annotation.Nonnull;
@@ -43,17 +53,63 @@
  * Abstract class that collects and bundles all write operations for {@link Transaction} and {@link
  * WriteBatch}.
  */
-public abstract class UpdateBuilder<T extends UpdateBuilder> {
+@InternalExtensionOnly
+public abstract class UpdateBuilder<T> {
+  static class WriteOperation {
+    Write.Builder write;
+    DocumentReference documentReference;
+
+    WriteOperation(DocumentReference documentReference, Write.Builder write) {
+      this.documentReference = documentReference;
+      this.write = write;
+    }
+  }
 
   final FirestoreImpl firestore;
-  private final List<Write.Builder> writes;
+  protected final List<WriteOperation> writes;
   private boolean committed;
+  private final int maxBatchSize;
+
+  protected BatchState state = BatchState.OPEN;
+  protected Map<DocumentReference, SettableApiFuture<WriteResult>> pendingOperations =
+      new HashMap<>();
+
+  /**
+   * Used to represent the state of batch.
+   *
+   * <p>Writes can only be added while the batch is OPEN. For a batch to be sent, the batch must be
+   * READY_TO_SEND. After a batch is sent, it is marked as SENT.
+   */
+  enum BatchState {
+    OPEN,
+    READY_TO_SEND,
+    SENT,
+  }
 
   UpdateBuilder(FirestoreImpl firestore) {
+    this(firestore, BulkWriter.MAX_BATCH_SIZE);
+  }
+
+  UpdateBuilder(FirestoreImpl firestore, int maxBatchSize) {
     this.firestore = firestore;
+    this.maxBatchSize = maxBatchSize;
     this.writes = new ArrayList<>();
   }
 
+  /**
+   * Wraps the result of the write operation before it is returned.
+   *
+   * <p>This method is used to generate the return value for all public methods. It allows
+   * operations on Transaction and Writebatch to return the object for chaining, while also allowing
+   * BulkWriter operations to return the future directly.
+   */
+  abstract T wrapResult(ApiFuture<WriteResult> result);
+
+  /** Whether to allow multiple writes to the same document in a batch. */
+  boolean allowDuplicateDocs() {
+    return true;
+  }
+
   /**
    * Turns a field map that contains field paths into a nested map. Turns {a.b : c} into {a : {b :
    * c}}.
@@ -124,9 +180,9 @@ private T performCreate(
       write.addAllUpdateTransforms(documentTransform.toPb());
     }
 
-    writes.add(write);
+    writes.add(new WriteOperation(documentReference, write));
 
-    return (T) this;
+    return wrapResult(processOperation(documentReference));
   }
 
   private void verifyNotCommitted() {
@@ -254,9 +310,9 @@ private T performSet(
       write.setUpdateMask(documentMask.toPb());
     }
 
-    writes.add(write);
+    writes.add(new WriteOperation(documentReference, write));
 
-    return (T) this;
+    return wrapResult(processOperation(documentReference));
   }
 
   /** Removes all values in 'fields' that are not specified in 'fieldMask'. */
@@ -339,18 +395,18 @@ public T update(
    *
    * @param documentReference The DocumentReference to update.
    * @param fields A Map containing the fields and values with which to update the document.
-   * @param options Preconditions to enforce on this update.
+   * @param precondition Precondition to enforce on this update.
    * @return The instance for chaining.
    */
   @Nonnull
   public T update(
       @Nonnull DocumentReference documentReference,
       @Nonnull Map<String, Object> fields,
-      Precondition options) {
+      Precondition precondition) {
     Preconditions.checkArgument(
-        !options.hasExists(), "Precondition 'exists' cannot be specified for update() calls.");
+        !precondition.hasExists(), "Precondition 'exists' cannot be specified for update() calls.");
     return performUpdate(
-        documentReference, convertToFieldPaths(fields, /* splitOnDots= */ true), options);
+        documentReference, convertToFieldPaths(fields, /* splitOnDots= */ true), precondition);
   }
 
   /**
@@ -402,7 +458,7 @@ public T update(
    * doesn't exist yet, the update will fail.
    *
    * @param documentReference The DocumentReference to update.
-   * @param options Preconditions to enforce on this update.
+   * @param precondition Precondition to enforce on this update.
    * @param field The first field to set.
    * @param value The first value to set.
    * @param moreFieldsAndValues String and Object pairs with more fields to be set.
@@ -411,15 +467,15 @@ public T update(
   @Nonnull
   public T update(
       @Nonnull DocumentReference documentReference,
-      @Nonnull Precondition options,
+      @Nonnull Precondition precondition,
       @Nonnull String field,
       @Nullable Object value,
       Object... moreFieldsAndValues) {
     Preconditions.checkArgument(
-        !options.hasExists(), "Precondition 'exists' cannot be specified for update() calls.");
+        !precondition.hasExists(), "Precondition 'exists' cannot be specified for update() calls.");
     return performUpdate(
         documentReference,
-        options,
+        precondition,
         FieldPath.fromDotSeparatedString(field),
         value,
         moreFieldsAndValues);
@@ -430,7 +486,7 @@ public T update(
    * doesn't exist yet, the update will fail.
    *
    * @param documentReference The DocumentReference to update.
-   * @param options Preconditions to enforce on this update.
+   * @param precondition Precondition to enforce on this update.
    * @param fieldPath The first field to set.
    * @param value The first value to set.
    * @param moreFieldsAndValues String and Object pairs with more fields to be set.
@@ -439,18 +495,18 @@ public T update(
   @Nonnull
   public T update(
       @Nonnull DocumentReference documentReference,
-      @Nonnull Precondition options,
+      @Nonnull Precondition precondition,
       @Nonnull FieldPath fieldPath,
       @Nullable Object value,
       Object... moreFieldsAndValues) {
     Preconditions.checkArgument(
-        !options.hasExists(), "Precondition 'exists' cannot be specified for update() calls.");
-    return performUpdate(documentReference, options, fieldPath, value, moreFieldsAndValues);
+        !precondition.hasExists(), "Precondition 'exists' cannot be specified for update() calls.");
+    return performUpdate(documentReference, precondition, fieldPath, value, moreFieldsAndValues);
   }
 
   private T performUpdate(
       @Nonnull DocumentReference documentReference,
-      @Nonnull Precondition options,
+      @Nonnull Precondition precondition,
       @Nonnull FieldPath fieldPath,
       @Nullable Object value,
       Object[] moreFieldsAndValues) {
@@ -483,7 +539,7 @@ private T performUpdate(
       fields.put(currentPath, objectValue);
     }
 
-    return performUpdate(documentReference, fields, options);
+    return performUpdate(documentReference, fields, precondition);
   }
 
   private T performUpdate(
@@ -523,9 +579,9 @@ public boolean allowTransform() {
     if (!documentTransform.isEmpty()) {
       write.addAllUpdateTransforms(documentTransform.toPb());
     }
-    writes.add(write);
+    writes.add(new WriteOperation(documentReference, write));
 
-    return (T) this;
+    return wrapResult(processOperation(documentReference));
   }
 
   /**
@@ -560,9 +616,9 @@ private T performDelete(
     if (!precondition.isEmpty()) {
       write.setCurrentDocument(precondition.toPb());
     }
-    writes.add(write);
+    writes.add(new WriteOperation(documentReference, write));
 
-    return (T) this;
+    return wrapResult(processOperation(documentReference));
   }
 
   /** Commit the current batch. */
@@ -576,8 +632,8 @@ ApiFuture<List<WriteResult>> commit(@Nullable ByteString transactionId) {
     final CommitRequest.Builder request = CommitRequest.newBuilder();
     request.setDatabase(firestore.getDatabaseName());
 
-    for (Write.Builder write : writes) {
-      request.addWrites(write);
+    for (WriteOperation writeOperation : writes) {
+      request.addWrites(writeOperation.write);
     }
 
     if (transactionId != null) {
@@ -609,6 +665,61 @@ public List<WriteResult> apply(CommitResponse commitResponse) {
         MoreExecutors.directExecutor());
   }
 
+  /**
+   * Commits all pending operations to the database and verifies all preconditions.
+   *
+   * <p>The writes in the batch are not applied atomically and can be applied out of order.
+   */
+  ApiFuture<List<BatchWriteResult>> bulkCommit() {
+    Tracing.getTracer()
+        .getCurrentSpan()
+        .addAnnotation(
+            "CloudFirestore.BatchWrite",
+            ImmutableMap.of("numDocuments", AttributeValue.longAttributeValue(writes.size())));
+
+    final BatchWriteRequest.Builder request = BatchWriteRequest.newBuilder();
+    request.setDatabase(firestore.getDatabaseName());
+
+    for (WriteOperation writeOperation : writes) {
+      request.addWrites(writeOperation.write);
+    }
+
+    ApiFuture<BatchWriteResponse> response =
+        firestore.sendRequest(request.build(), firestore.getClient().batchWriteCallable());
+
+    return ApiFutures.transform(
+        response,
+        new ApiFunction<BatchWriteResponse, List<BatchWriteResult>>() {
+          @Override
+          public List<BatchWriteResult> apply(BatchWriteResponse batchWriteResponse) {
+            List<com.google.firestore.v1.WriteResult> writeResults =
+                batchWriteResponse.getWriteResultsList();
+
+            List<com.google.rpc.Status> statuses = batchWriteResponse.getStatusList();
+
+            List<BatchWriteResult> result = new ArrayList<>();
+
+            for (int i = 0; i < writeResults.size(); ++i) {
+              com.google.firestore.v1.WriteResult writeResult = writeResults.get(i);
+              com.google.rpc.Status status = statuses.get(i);
+              Status code = Status.fromCodeValue(status.getCode());
+              @Nullable Timestamp updateTime = null;
+              @Nullable Exception exception = null;
+              if (code == Status.OK) {
+                updateTime = Timestamp.fromProto(writeResult.getUpdateTime());
+              } else {
+                exception = FirestoreException.serverRejected(code, status.getMessage());
+              }
+              result.add(
+                  new BatchWriteResult(writes.get(i).documentReference, updateTime, exception));
+            }
+
+            return result;
+          }
+        },
+        MoreExecutors.directExecutor());
+  }
+
   /** Checks whether any updates have been queued. */
   boolean isEmpty() {
     return writes.isEmpty();
@@ -618,4 +729,87 @@ boolean isEmpty() {
   public int getMutationsSize() {
     return writes.size();
   }
+
+  BatchState getState() {
+    return state;
+  }
+
+  boolean hasDocument(DocumentReference documentReference) {
+    return pendingOperations.containsKey(documentReference);
+  }
+
+  Set<DocumentReference> getPendingDocuments() {
+    return pendingOperations.keySet();
+  }
+
+  Collection<SettableApiFuture<WriteResult>> getPendingFutures() {
+    return pendingOperations.values();
+  }
+
+  int getPendingOperationCount() {
+    return pendingOperations.size();
+  }
+
+  private ApiFuture<WriteResult> processOperation(DocumentReference documentReference) {
+    Preconditions.checkState(
+        allowDuplicateDocs() || !pendingOperations.containsKey(documentReference),
+        "Batch should not contain writes to the same document");
+    Preconditions.checkState(state == BatchState.OPEN, "Batch should be OPEN when adding writes");
+    SettableApiFuture<WriteResult> result = SettableApiFuture.create();
+    pendingOperations.put(documentReference, result);
+
+    if (getPendingOperationCount() == maxBatchSize) {
+      state = BatchState.READY_TO_SEND;
+    }
+
+    return result;
+  }
+
+  /**
+   * Resolves the individual operations in the batch with the results and removes the entry from the
+   * pendingOperations map if the result is not retryable.
+   */
+  void processResults(List<BatchWriteResult> results) {
+    for (BatchWriteResult result : results) {
+      if (result.getException() == null || !shouldRetry(result.getException())) {
+        convertBatchWriteResult(result, pendingOperations.get(result.getDocumentReference()));
+        pendingOperations.remove(result.getDocumentReference());
+      }
+    }
+  }
+
+  void failRemainingOperations(List<BatchWriteResult> results) {
+    for (BatchWriteResult result : results) {
+      convertBatchWriteResult(result, pendingOperations.get(result.getDocumentReference()));
+    }
+  }
+
+  private boolean shouldRetry(Exception exception) {
+    if (!(exception instanceof FirestoreException)) {
+      return false;
+    }
+    Set<Code> codes = FirestoreSettings.newBuilder().batchWriteSettings().getRetryableCodes();
+    Status status = ((FirestoreException) exception).getStatus();
+    for (Code code : codes) {
+      if (code.equals(Code.valueOf(status.getCode().name()))) {
+        return true;
+      }
+    }
+    return false;
+  }
+
+  private void convertBatchWriteResult(
+      BatchWriteResult result, SettableApiFuture<WriteResult> future) {
+    if (result.getWriteTime() != null) {
+      future.set(new WriteResult(result.getWriteTime()));
+    } else {
+      future.setException(result.getException());
+    }
+  }
+
+  void markReadyToSend() {
+    if (state == BatchState.OPEN) {
+      state = BatchState.READY_TO_SEND;
+    }
+  }
 }
diff --git a/google-cloud-firestore/src/main/java/com/google/cloud/firestore/WriteBatch.java b/google-cloud-firestore/src/main/java/com/google/cloud/firestore/WriteBatch.java
index 816bf41e4..ba69ab420 100644
--- a/google-cloud-firestore/src/main/java/com/google/cloud/firestore/WriteBatch.java
+++ b/google-cloud-firestore/src/main/java/com/google/cloud/firestore/WriteBatch.java
@@ -40,4 +40,8 @@ public class WriteBatch extends UpdateBuilder<WriteBatch> {
   public ApiFuture<List<WriteResult>> commit() {
     return super.commit(null);
   }
+
+  WriteBatch wrapResult(ApiFuture<WriteResult> result) {
+    return this;
+  }
 }
diff --git a/google-cloud-firestore/src/main/java/com/google/cloud/firestore/WriteResult.java b/google-cloud-firestore/src/main/java/com/google/cloud/firestore/WriteResult.java
index 12106b835..a8abc4901 100644
--- a/google-cloud-firestore/src/main/java/com/google/cloud/firestore/WriteResult.java
+++ b/google-cloud-firestore/src/main/java/com/google/cloud/firestore/WriteResult.java
@@ -25,7 +25,7 @@ public final class WriteResult {
 
   private final Timestamp updateTime;
 
-  private WriteResult(Timestamp updateTime) {
+  WriteResult(Timestamp updateTime) {
     this.updateTime = updateTime;
   }
 
diff --git a/google-cloud-firestore/src/main/java/com/google/cloud/firestore/spi/v1/FirestoreRpc.java b/google-cloud-firestore/src/main/java/com/google/cloud/firestore/spi/v1/FirestoreRpc.java
index 29bb676f5..d31d6557c 100644
--- a/google-cloud-firestore/src/main/java/com/google/cloud/firestore/spi/v1/FirestoreRpc.java
+++ b/google-cloud-firestore/src/main/java/com/google/cloud/firestore/spi/v1/FirestoreRpc.java
@@ -25,6 +25,8 @@
 import com.google.cloud.firestore.v1.FirestoreClient.ListDocumentsPagedResponse;
 import com.google.firestore.v1.BatchGetDocumentsRequest;
 import com.google.firestore.v1.BatchGetDocumentsResponse;
+import com.google.firestore.v1.BatchWriteRequest;
+import com.google.firestore.v1.BatchWriteResponse;
 import com.google.firestore.v1.BeginTransactionRequest;
 import com.google.firestore.v1.BeginTransactionResponse;
 import com.google.firestore.v1.CommitRequest;
@@ -46,6 +48,8 @@ public interface FirestoreRpc extends AutoCloseable, ServiceRpc {
   /** Commits a transaction, while optionally updating documents. */
   UnaryCallable<CommitRequest, CommitResponse> commitCallable();
 
+  UnaryCallable<BatchWriteRequest, BatchWriteResponse> batchWriteCallable();
+
   /** Gets multiple documents. */
   ServerStreamingCallable<BatchGetDocumentsRequest, BatchGetDocumentsResponse>
       batchGetDocumentsCallable();
diff --git a/google-cloud-firestore/src/main/java/com/google/cloud/firestore/spi/v1/GrpcFirestoreRpc.java b/google-cloud-firestore/src/main/java/com/google/cloud/firestore/spi/v1/GrpcFirestoreRpc.java
index 2eccae513..d99e098a2 100644
--- a/google-cloud-firestore/src/main/java/com/google/cloud/firestore/spi/v1/GrpcFirestoreRpc.java
+++ b/google-cloud-firestore/src/main/java/com/google/cloud/firestore/spi/v1/GrpcFirestoreRpc.java
@@ -44,6 +44,8 @@
 import com.google.cloud.grpc.GrpcTransportOptions.ExecutorFactory;
 import com.google.firestore.v1.BatchGetDocumentsRequest;
 import com.google.firestore.v1.BatchGetDocumentsResponse;
+import com.google.firestore.v1.BatchWriteRequest;
+import com.google.firestore.v1.BatchWriteResponse;
 import com.google.firestore.v1.BeginTransactionRequest;
 import com.google.firestore.v1.BeginTransactionResponse;
 import com.google.firestore.v1.CommitRequest;
@@ -163,6 +165,11 @@ public UnaryCallable<CommitRequest, CommitResponse> commitCallable() {
     return firestoreStub.commitCallable();
   }
 
+  @Override
+  public UnaryCallable<BatchWriteRequest, BatchWriteResponse> batchWriteCallable() {
+    return firestoreStub.batchWriteCallable();
+  }
+
   @Override
   public ServerStreamingCallable<BatchGetDocumentsRequest, BatchGetDocumentsResponse>
       batchGetDocumentsCallable() {
diff --git a/google-cloud-firestore/src/test/java/com/google/cloud/firestore/BulkWriterTest.java b/google-cloud-firestore/src/test/java/com/google/cloud/firestore/BulkWriterTest.java
new file mode 100644
index 000000000..e5f72af65
--- /dev/null
+++ b/google-cloud-firestore/src/test/java/com/google/cloud/firestore/BulkWriterTest.java
@@ -0,0 +1,758 @@
+/*
+ * Copyright 2020 Google LLC
+ *
+ * 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.firestore;
+
+import static com.google.cloud.firestore.LocalFirestoreHelper.batchWrite;
+import static com.google.cloud.firestore.LocalFirestoreHelper.create;
+import static com.google.cloud.firestore.LocalFirestoreHelper.delete;
+import static com.google.cloud.firestore.LocalFirestoreHelper.set;
+import static com.google.cloud.firestore.LocalFirestoreHelper.update;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doReturn;
+
+import com.google.api.core.ApiFuture;
+import com.google.api.core.ApiFutures;
+import com.google.api.core.SettableApiFuture;
+import com.google.api.gax.rpc.UnaryCallable;
+import com.google.cloud.Timestamp;
+import com.google.cloud.firestore.LocalFirestoreHelper.ResponseStubber;
+import com.google.cloud.firestore.LocalFirestoreHelper.SerialResponseStubber;
+import com.google.cloud.firestore.spi.v1.FirestoreRpc;
+import com.google.firestore.v1.BatchWriteRequest;
+import com.google.firestore.v1.BatchWriteResponse;
+import com.google.protobuf.GeneratedMessageV3;
+import com.google.rpc.Code;
+import io.grpc.Status;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import javax.annotation.Nonnull;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.Timeout;
+import org.junit.runner.RunWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Captor;
+import org.mockito.Matchers;
+import org.mockito.Mockito;
+import org.mockito.Spy;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.runners.MockitoJUnitRunner;
+import org.mockito.stubbing.Answer;
+
+@RunWith(MockitoJUnitRunner.class)
+public class BulkWriterTest {
+
+  @Rule public Timeout timeout = new Timeout(500, TimeUnit.MILLISECONDS);
+
+  @Spy private final FirestoreRpc firestoreRpc = Mockito.mock(FirestoreRpc.class);
+
+  /** Executor that executes delayed tasks without delay. */
+  private final ScheduledExecutorService immediateExecutor =
+      new ScheduledThreadPoolExecutor(1) {
+        @Override
+        @Nonnull
+        public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) {
+          return super.schedule(command, 0, TimeUnit.MILLISECONDS);
+        }
+      };
+
+  @Spy
+  private final FirestoreImpl firestoreMock =
+      new FirestoreImpl(
+          FirestoreOptions.newBuilder().setProjectId("test-project").build(), firestoreRpc);
+
+  @Captor private ArgumentCaptor<BatchWriteRequest> batchWriteCapture;
+
+  private BulkWriter bulkWriter;
+  private DocumentReference doc1;
+  private DocumentReference doc2;
+
+  private ApiFuture<BatchWriteResponse> successResponse(int updateTimeSeconds) {
+    BatchWriteResponse.Builder response = BatchWriteResponse.newBuilder();
+    response.addWriteResultsBuilder().getUpdateTimeBuilder().setSeconds(updateTimeSeconds).build();
+    response.addStatusBuilder().build();
+    return ApiFutures.immediateFuture(response.build());
+  }
+
+  private ApiFuture<BatchWriteResponse> failedResponse(int code) {
+    BatchWriteResponse.Builder response = BatchWriteResponse.newBuilder();
+    response.addWriteResultsBuilder().build();
+    response.addStatusBuilder().setCode(code).build();
+    return ApiFutures.immediateFuture(response.build());
+  }
+
+  private ApiFuture<BatchWriteResponse> failedResponse() {
+    return failedResponse(Code.DEADLINE_EXCEEDED_VALUE);
+  }
+
+  private ApiFuture<BatchWriteResponse> mergeResponses(ApiFuture<BatchWriteResponse>... responses)
+      throws Exception {
+    BatchWriteResponse.Builder response = BatchWriteResponse.newBuilder();
+    for (ApiFuture<BatchWriteResponse> future : responses) {
+      BatchWriteResponse res = future.get();
+      response.addStatus(res.getStatus(0));
+      response.addWriteResults(res.getWriteResults(0));
+    }
+    return ApiFutures.immediateFuture(response.build());
+  }
+
+  private void verifyRequests(List<BatchWriteRequest> requests, ResponseStubber responseStubber) {
+    int index = 0;
+    for (GeneratedMessageV3 request : responseStubber.keySet()) {
+      assertEquals(request, requests.get(index++));
+    }
+  }
+
+  @Before
+  public void before() {
+    doReturn(immediateExecutor).when(firestoreRpc).getExecutor();
+    bulkWriter = firestoreMock.bulkWriter();
+    doc1 = firestoreMock.document("coll/doc1");
+    doc2 = firestoreMock.document("coll/doc2");
+  }
+
+  @Test
+  public void hasSetMethod() throws Exception {
+    ResponseStubber responseStubber =
+        new ResponseStubber() {
+          {
+            put(
+                batchWrite(set(LocalFirestoreHelper.SINGLE_FIELD_PROTO, "coll/doc1")),
+                successResponse(2));
+          }
+        };
+    responseStubber.initializeStub(batchWriteCapture, firestoreMock);
+
+    ApiFuture<WriteResult> result = bulkWriter.set(doc1, LocalFirestoreHelper.SINGLE_FIELD_MAP);
+    bulkWriter.close();
+
+    List<BatchWriteRequest> requests = batchWriteCapture.getAllValues();
+    assertEquals(responseStubber.size(), requests.size());
+
+    verifyRequests(requests, responseStubber);
+    assertEquals(Timestamp.ofTimeSecondsAndNanos(2, 0), result.get().getUpdateTime());
+  }
+
+  @Test
+  public void hasUpdateMethod() throws Exception {
+    ResponseStubber responseStubber =
+        new ResponseStubber() {
+          {
+            put(
+                batchWrite(update(LocalFirestoreHelper.SINGLE_FIELD_PROTO, "coll/doc1")),
+                successResponse(2));
+          }
+        };
+    responseStubber.initializeStub(batchWriteCapture, firestoreMock);
+
+    ApiFuture<WriteResult> result = bulkWriter.update(doc1, LocalFirestoreHelper.SINGLE_FIELD_MAP);
+    bulkWriter.close();
+
+    List<BatchWriteRequest> requests = batchWriteCapture.getAllValues();
+    assertEquals(responseStubber.size(), requests.size());
+
+    verifyRequests(requests, responseStubber);
+    assertEquals(Timestamp.ofTimeSecondsAndNanos(2, 0), result.get().getUpdateTime());
+  }
+
+  @Test
+  public void hasDeleteMethod() throws Exception {
+    ResponseStubber responseStubber =
+        new ResponseStubber() {
+          {
+            put(batchWrite(delete("coll/doc1")), successResponse(2));
+          }
+        };
+    responseStubber.initializeStub(batchWriteCapture, firestoreMock);
+
+    ApiFuture<WriteResult> result = bulkWriter.delete(doc1);
+    bulkWriter.close();
+
+    List<BatchWriteRequest> requests = batchWriteCapture.getAllValues();
+    assertEquals(responseStubber.size(), requests.size());
+
+    verifyRequests(requests, responseStubber);
+    assertEquals(Timestamp.ofTimeSecondsAndNanos(2, 0), result.get().getUpdateTime());
+  }
+
+  @Test
+  public void hasCreateMethod() throws Exception {
+    ResponseStubber responseStubber =
+        new ResponseStubber() {
+          {
+            put(
+                batchWrite(create(LocalFirestoreHelper.SINGLE_FIELD_PROTO, "coll/doc1")),
+                successResponse(2));
+          }
+        };
+    responseStubber.initializeStub(batchWriteCapture, firestoreMock);
+
+    ApiFuture<WriteResult> result = bulkWriter.create(doc1, LocalFirestoreHelper.SINGLE_FIELD_MAP);
+    bulkWriter.close();
+
+    List<BatchWriteRequest> requests = batchWriteCapture.getAllValues();
+    assertEquals(responseStubber.size(), requests.size());
+
+    verifyRequests(requests, responseStubber);
+    assertEquals(Timestamp.ofTimeSecondsAndNanos(2, 0), result.get().getUpdateTime());
+  }
+
+  @Test
+  public void surfacesErrors() throws Exception {
+    ResponseStubber responseStubber =
+        new ResponseStubber() {
+          {
+            put(
+                batchWrite(set(LocalFirestoreHelper.SINGLE_FIELD_PROTO, "coll/doc1")),
+                failedResponse());
+          }
+        };
+    responseStubber.initializeStub(batchWriteCapture, firestoreMock);
+
+    ApiFuture<WriteResult> result = bulkWriter.set(doc1, LocalFirestoreHelper.SINGLE_FIELD_MAP);
+    bulkWriter.close();
+
+    List<BatchWriteRequest> requests = batchWriteCapture.getAllValues();
+    assertEquals(responseStubber.size(), requests.size());
+
+    verifyRequests(requests, responseStubber);
+    try {
+      result.get();
+      fail("set() should have failed");
+    } catch (Exception e) {
+      assertTrue(e.getCause() instanceof FirestoreException);
+      assertEquals(Status.DEADLINE_EXCEEDED, ((FirestoreException) e.getCause()).getStatus());
+    }
+  }
+
+  @Test
+  public void flushResolvesImmediatelyIfNoWrites() throws Exception {
+    bulkWriter.flush().get();
+  }
+
+  @Test
+  public void addsWritesToNewBatchAfterFlush() throws Exception {
+    ResponseStubber responseStubber =
+        new ResponseStubber() {
+          {
+            put(
+                batchWrite(create(LocalFirestoreHelper.SINGLE_FIELD_PROTO, "coll/doc1")),
+                successResponse(1));
+            put(
+                batchWrite(set(LocalFirestoreHelper.SINGLE_FIELD_PROTO, "coll/doc2")),
+                successResponse(2));
+          }
+        };
+    responseStubber.initializeStub(batchWriteCapture, firestoreMock);
+
+    ApiFuture<WriteResult> result1 = bulkWriter.create(doc1, LocalFirestoreHelper.SINGLE_FIELD_MAP);
+    bulkWriter.flush();
+    ApiFuture<WriteResult> result2 = bulkWriter.set(doc2, LocalFirestoreHelper.SINGLE_FIELD_MAP);
+    bulkWriter.close();
+
+    List<BatchWriteRequest> requests = batchWriteCapture.getAllValues();
+    assertEquals(responseStubber.size(), requests.size());
+
+    verifyRequests(requests, responseStubber);
+    assertEquals(Timestamp.ofTimeSecondsAndNanos(1, 0), result1.get().getUpdateTime());
+    assertEquals(Timestamp.ofTimeSecondsAndNanos(2, 0), result2.get().getUpdateTime());
+  }
+
+  @Test
+  public void closeResolvesImmediatelyIfNoWrites() throws Exception {
+    bulkWriter.close();
+  }
+
+  @Test
+  public void cannotCallMethodsAfterClose() throws Exception {
+    String expected = "BulkWriter has already been closed.";
+    bulkWriter.close();
+    try {
+      bulkWriter.set(doc1, LocalFirestoreHelper.SINGLE_FIELD_MAP);
+      fail("set() should have failed");
+    } catch (Exception e) {
+      assertEquals(expected, e.getMessage());
+    }
+    try {
+      bulkWriter.create(doc1, LocalFirestoreHelper.SINGLE_FIELD_MAP);
+      fail("create() should have failed");
+    } catch (Exception e) {
+      assertEquals(expected, e.getMessage());
+    }
+    try {
+      bulkWriter.update(doc1, LocalFirestoreHelper.SINGLE_FIELD_MAP);
+      fail("update() should have failed");
+    } catch (Exception e) {
+      assertEquals(expected, e.getMessage());
+    }
+    try {
+      bulkWriter.delete(doc1);
+      fail("delete() should have failed");
+    } catch (Exception e) {
+      assertEquals(expected, e.getMessage());
+    }
+    try {
+      bulkWriter.flush();
+      fail("flush() should have failed");
+    } catch (Exception e) {
+      assertEquals(expected, e.getMessage());
+    }
+    try {
+      bulkWriter.close();
+      fail("close() should have failed");
+    } catch (Exception e) {
+      assertEquals(expected, e.getMessage());
+    }
+  }
+
+  @Test
+  public void sendsWritesToSameDocInSeparateBatches() throws Exception {
+    ResponseStubber responseStubber =
+        new ResponseStubber() {
+          {
+            put(
+                batchWrite(set(LocalFirestoreHelper.SINGLE_FIELD_PROTO, "coll/doc1")),
+                successResponse(1));
+            put(
+                batchWrite(update(LocalFirestoreHelper.UPDATED_FIELD_PROTO, "coll/doc1")),
+                successResponse(2));
+          }
+        };
+    responseStubber.initializeStub(batchWriteCapture, firestoreMock);
+
+    // Create another document reference pointing to the same document.
+    DocumentReference sameDoc = firestoreMock.document(doc1.getPath());
+    ApiFuture<WriteResult> result1 = bulkWriter.set(doc1, LocalFirestoreHelper.SINGLE_FIELD_MAP);
+    ApiFuture<WriteResult> result2 =
+        bulkWriter.update(sameDoc, LocalFirestoreHelper.UPDATED_FIELD_MAP);
+    bulkWriter.close();
+
+    List<BatchWriteRequest> requests = batchWriteCapture.getAllValues();
+    assertEquals(responseStubber.size(), requests.size());
+
+    verifyRequests(requests, responseStubber);
+    assertEquals(Timestamp.ofTimeSecondsAndNanos(1, 0), result1.get().getUpdateTime());
+    assertEquals(Timestamp.ofTimeSecondsAndNanos(2, 0), result2.get().getUpdateTime());
+  }
+
+  @Test
+  public void sendWritesToDifferentDocsInSameBatch() throws Exception {
+    ResponseStubber responseStubber =
+        new ResponseStubber() {
+          {
+            put(
+                batchWrite(
+                    set(LocalFirestoreHelper.SINGLE_FIELD_PROTO, "coll/doc1"),
+                    update(LocalFirestoreHelper.SINGLE_FIELD_PROTO, "coll/doc2")),
+                mergeResponses(successResponse(1), successResponse(2)));
+          }
+        };
+    responseStubber.initializeStub(batchWriteCapture, firestoreMock);
+
+    ApiFuture<WriteResult> result1 = bulkWriter.set(doc1, LocalFirestoreHelper.SINGLE_FIELD_MAP);
+    ApiFuture<WriteResult> result2 = bulkWriter.update(doc2, LocalFirestoreHelper.SINGLE_FIELD_MAP);
+    bulkWriter.close();
+
+    List<BatchWriteRequest> requests = batchWriteCapture.getAllValues();
+    assertEquals(responseStubber.size(), requests.size());
+
+    verifyRequests(requests, responseStubber);
+    assertEquals(Timestamp.ofTimeSecondsAndNanos(1, 0), result1.get().getUpdateTime());
+    assertEquals(Timestamp.ofTimeSecondsAndNanos(2, 0), result2.get().getUpdateTime());
+  }
+
+  @Test
+  public void sendBatchesWhenSizeLimitIsReached() throws Exception {
+    ResponseStubber responseStubber =
+        new ResponseStubber() {
+          {
+            put(
+                batchWrite(
+                    set(LocalFirestoreHelper.SINGLE_FIELD_PROTO, "coll/doc1"),
+                    update(LocalFirestoreHelper.SINGLE_FIELD_PROTO, "coll/doc2"),
+                    create(LocalFirestoreHelper.SINGLE_FIELD_PROTO, "coll/doc3")),
+                mergeResponses(successResponse(1), successResponse(2), successResponse(3)));
+          }
+        };
+    responseStubber.initializeStub(batchWriteCapture, firestoreMock);
+
+    bulkWriter.setMaxBatchSize(3);
+    ApiFuture<WriteResult> result1 = bulkWriter.set(doc1, LocalFirestoreHelper.SINGLE_FIELD_MAP);
+    ApiFuture<WriteResult> result2 = bulkWriter.update(doc2, LocalFirestoreHelper.SINGLE_FIELD_MAP);
+    ApiFuture<WriteResult> result3 =
+        bulkWriter.create(
+            firestoreMock.document("coll/doc3"), LocalFirestoreHelper.SINGLE_FIELD_MAP);
+
+    // The 4th write should not be sent because it should be in a new batch.
+    bulkWriter.delete(firestoreMock.document("coll/doc4"));
+
+    assertEquals(Timestamp.ofTimeSecondsAndNanos(1, 0), result1.get().getUpdateTime());
+    assertEquals(Timestamp.ofTimeSecondsAndNanos(2, 0), result2.get().getUpdateTime());
+    assertEquals(Timestamp.ofTimeSecondsAndNanos(3, 0), result3.get().getUpdateTime());
+
+    List<BatchWriteRequest> requests = batchWriteCapture.getAllValues();
+    assertEquals(responseStubber.size(), requests.size());
+    verifyRequests(requests, responseStubber);
+  }
+
+  @Test
+  public void retriesIndividualWritesThatFailWithAbortedOrUnavailable() throws Exception {
+    ResponseStubber responseStubber =
+        new ResponseStubber() {
+          {
+            put(
+                batchWrite(
+                    set(LocalFirestoreHelper.SINGLE_FIELD_PROTO, "coll/doc1"),
+                    set(LocalFirestoreHelper.SINGLE_FIELD_PROTO, "coll/doc2"),
+                    set(LocalFirestoreHelper.SINGLE_FIELD_PROTO, "coll/doc3")),
+                mergeResponses(
+                    failedResponse(),
+                    failedResponse(Code.UNAVAILABLE_VALUE),
+                    failedResponse(Code.ABORTED_VALUE)));
+            put(
+                batchWrite(
+                    set(LocalFirestoreHelper.SINGLE_FIELD_PROTO, "coll/doc2"),
+                    set(LocalFirestoreHelper.SINGLE_FIELD_PROTO, "coll/doc3")),
+                mergeResponses(successResponse(2), failedResponse(Code.ABORTED_VALUE)));
+            put(
+                batchWrite(set(LocalFirestoreHelper.SINGLE_FIELD_PROTO, "coll/doc3")),
+                successResponse(3));
+          }
+        };
+    responseStubber.initializeStub(batchWriteCapture, firestoreMock);
+
+    ApiFuture<WriteResult> result1 = bulkWriter.set(doc1, LocalFirestoreHelper.SINGLE_FIELD_MAP);
+    ApiFuture<WriteResult> result2 = bulkWriter.set(doc2, LocalFirestoreHelper.SINGLE_FIELD_MAP);
+    ApiFuture<WriteResult> result3 =
+        bulkWriter.set(firestoreMock.document("coll/doc3"), LocalFirestoreHelper.SINGLE_FIELD_MAP);
+    bulkWriter.close();
+
+    try {
+      result1.get();
+      fail("set() should have failed");
+    } catch (Exception e) {
+      assertTrue(e.getCause() instanceof FirestoreException);
+      assertEquals(Status.DEADLINE_EXCEEDED, ((FirestoreException) e.getCause()).getStatus());
+    }
+    assertEquals(Timestamp.ofTimeSecondsAndNanos(2, 0), result2.get().getUpdateTime());
+    assertEquals(Timestamp.ofTimeSecondsAndNanos(3, 0), result3.get().getUpdateTime());
+
+    List<BatchWriteRequest> requests = batchWriteCapture.getAllValues();
+    assertEquals(responseStubber.size(), requests.size());
+  }
+
+  @Test
+  public void writesCompleteInCorrectOrderBeforeFlush() throws Exception {
+    ResponseStubber responseStubber =
+        new ResponseStubber() {
+          {
+            put(
+                batchWrite(
+                    set(LocalFirestoreHelper.SINGLE_FIELD_PROTO, "coll/doc1"),
+                    set(LocalFirestoreHelper.SINGLE_FIELD_PROTO, "coll/doc2")),
+                mergeResponses(successResponse(1), failedResponse(Code.ABORTED_VALUE)));
+            put(
+                batchWrite(set(LocalFirestoreHelper.SINGLE_FIELD_PROTO, "coll/doc2")),
+                successResponse(2));
+          }
+        };
+    responseStubber.initializeStub(batchWriteCapture, firestoreMock);
+    final List<String> completions = new ArrayList<>();
+    ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
+    final SettableApiFuture<Void> flushComplete = SettableApiFuture.create();
+
+    bulkWriter
+        .set(doc1, LocalFirestoreHelper.SINGLE_FIELD_MAP)
+        .addListener(
+            new Runnable() {
+              public void run() {
+                completions.add("doc1");
+              }
+            },
+            executor);
+    bulkWriter
+        .set(doc2, LocalFirestoreHelper.SINGLE_FIELD_MAP)
+        .addListener(
+            new Runnable() {
+              public void run() {
+                completions.add("doc2");
+              }
+            },
+            executor);
+    ;
+
+    ApiFuture<Void> flush = bulkWriter.flush();
+    flush.addListener(
+        new Runnable() {
+          public void run() {
+            completions.add("flush");
+            flushComplete.set(null);
+          }
+        },
+        executor);
+
+    flushComplete.get();
+    assertEquals("doc1", completions.get(0));
+    assertEquals("doc2", completions.get(1));
+    assertEquals("flush", completions.get(2));
+  }
+
+  @Test
+  public void flushCompletesWhenAllWritesComplete() throws Exception {
+    ResponseStubber responseStubber =
+        new ResponseStubber() {
+          {
+            put(
+                batchWrite(
+                    set(LocalFirestoreHelper.SINGLE_FIELD_PROTO, "coll/doc1"),
+                    set(LocalFirestoreHelper.SINGLE_FIELD_PROTO, "coll/doc2"),
+                    set(LocalFirestoreHelper.SINGLE_FIELD_PROTO, "coll/doc3")),
+                mergeResponses(failedResponse(), successResponse(1), successResponse(1)));
+          }
+        };
+    responseStubber.initializeStub(batchWriteCapture, firestoreMock);
+
+    ApiFuture<WriteResult> result1 = bulkWriter.set(doc1, LocalFirestoreHelper.SINGLE_FIELD_MAP);
+    ApiFuture<WriteResult> result2 = bulkWriter.set(doc2, LocalFirestoreHelper.SINGLE_FIELD_MAP);
+    ApiFuture<WriteResult> result3 =
+        bulkWriter.set(firestoreMock.document("coll/doc3"), LocalFirestoreHelper.SINGLE_FIELD_MAP);
+    bulkWriter.close();
+    assertTrue(result1.isDone());
+    assertTrue(result2.isDone());
+    assertTrue(result3.isDone());
+  }
+
+  @Test
+  public void doesNotSendBatchesIfSameDocIsInFlight() throws Exception {
+    final SerialResponseStubber responseStubber =
+        new SerialResponseStubber() {
+          {
+            put(
+                batchWrite(
+                    set(LocalFirestoreHelper.SINGLE_FIELD_PROTO, "coll/doc1"),
+                    set(LocalFirestoreHelper.SINGLE_FIELD_PROTO, "coll/doc2")),
+                mergeResponses(successResponse(1), successResponse(2)));
+            put(
+                batchWrite(set(LocalFirestoreHelper.SINGLE_FIELD_PROTO, "coll/doc1")),
+                successResponse(3));
+          }
+        };
+    responseStubber.initializeStub(batchWriteCapture, firestoreMock);
+    bulkWriter.set(doc1, LocalFirestoreHelper.SINGLE_FIELD_MAP);
+    bulkWriter.set(doc2, LocalFirestoreHelper.SINGLE_FIELD_MAP);
+
+    // Schedule flush on separate thread to avoid blocking main thread while waiting for
+    // activeRequestComplete.
+    ScheduledFuture<ApiFuture<Void>> flush1 =
+        Executors.newSingleThreadScheduledExecutor()
+            .schedule(
+                new Callable<ApiFuture<Void>>() {
+                  public ApiFuture<Void> call() {
+                    return bulkWriter.flush();
+                  }
+                },
+                0,
+                TimeUnit.MILLISECONDS);
+
+    // Wait for flush() to perform logic and reach the stubbed response. This simulates a first
+    // batch that has been sent with its response still pending.
+    responseStubber.awaitRequest();
+
+    bulkWriter.set(doc1, LocalFirestoreHelper.SINGLE_FIELD_MAP);
+    ApiFuture<Void> flush2 = bulkWriter.flush();
+
+    // Wait for flush() to receive its response and process the batch.
+    responseStubber.markAllRequestsComplete();
+    flush1.get().get();
+    flush2.get();
+    bulkWriter.close();
+
+    List<BatchWriteRequest> requests = batchWriteCapture.getAllValues();
+    assertEquals(responseStubber.size(), requests.size());
+
+    verifyRequests(requests, responseStubber);
+  }
+
+  @Test
+  public void doesNotSendBatchesIfDoingSoExceedsRateLimit() {
+    final boolean[] timeoutCalled = {false};
+    final ScheduledExecutorService timeoutExecutor =
+        new ScheduledThreadPoolExecutor(1) {
+          @Override
+          @Nonnull
+          public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) {
+            if (delay > 0) {
+              timeoutCalled[0] = true;
+            }
+            return super.schedule(command, 0, TimeUnit.MILLISECONDS);
+          }
+        };
+    doReturn(timeoutExecutor).when(firestoreRpc).getExecutor();
+    BulkWriter bulkWriter = firestoreMock.bulkWriter();
+
+    for (int i = 0; i < 600; ++i) {
+      bulkWriter.set(firestoreMock.document("coll/doc" + i), LocalFirestoreHelper.SINGLE_FIELD_MAP);
+    }
+    bulkWriter.flush();
+
+    assertTrue(timeoutCalled[0]);
+  }
+
+  @Test
+  public void flushSucceedsEvenIfBulkCommitFails() throws Exception {
+    doReturn(
+            ApiFutures.immediateFailedFuture(
+                new IllegalStateException("Mock batchWrite failed in test")))
+        .when(firestoreMock)
+        .sendRequest(
+            batchWriteCapture.capture(),
+            Matchers.<UnaryCallable<BatchWriteRequest, BatchWriteResponse>>any());
+    bulkWriter.set(doc1, LocalFirestoreHelper.SINGLE_FIELD_MAP);
+    bulkWriter.set(doc2, LocalFirestoreHelper.SINGLE_FIELD_MAP);
+    bulkWriter.flush().get();
+  }
+
+  @Test
+  public void closeSucceedsEvenIfBulkCommitFails() throws Exception {
+    doReturn(
+            ApiFutures.immediateFailedFuture(
+                new IllegalStateException("Mock batchWrite failed in test")))
+        .when(firestoreMock)
+        .sendRequest(
+            batchWriteCapture.capture(),
+            Matchers.<UnaryCallable<BatchWriteRequest, BatchWriteResponse>>any());
+    bulkWriter.set(doc1, LocalFirestoreHelper.SINGLE_FIELD_MAP);
+    bulkWriter.set(doc2, LocalFirestoreHelper.SINGLE_FIELD_MAP);
+    bulkWriter.close();
+  }
+
+  @Test
+  public void individualWritesErrorIfBulkCommitFails() throws Exception {
+    doReturn(
+            ApiFutures.immediateFailedFuture(
+                FirestoreException.serverRejected(
+                    Status.DEADLINE_EXCEEDED, "Mock batchWrite failed in test")))
+        .when(firestoreMock)
+        .sendRequest(
+            batchWriteCapture.capture(),
+            Matchers.<UnaryCallable<BatchWriteRequest, BatchWriteResponse>>any());
+    int opCount = 0;
+    ApiFuture<WriteResult> result1 = bulkWriter.set(doc1, LocalFirestoreHelper.SINGLE_FIELD_MAP);
+    ApiFuture<WriteResult> result2 = bulkWriter.set(doc2, LocalFirestoreHelper.SINGLE_FIELD_MAP);
+    bulkWriter.close();
+
+    for (ApiFuture<WriteResult> result : Arrays.asList(result1, result2)) {
+      try {
+        result.get();
+      } catch (Exception e) {
+        assertTrue(e.getMessage().contains("Mock batchWrite failed in test"));
+        ++opCount;
+      }
+    }
+    assertEquals(2, opCount);
+  }
+
+  @Test
+  public void individualWritesErrorIfBulkCommitFailsWithNonFirestoreException() throws Exception {
+    doReturn(
+            ApiFutures.immediateFailedFuture(
+                new IllegalStateException("Mock batchWrite failed in test")))
+        .when(firestoreMock)
+        .sendRequest(
+            batchWriteCapture.capture(),
+            Matchers.<UnaryCallable<BatchWriteRequest, BatchWriteResponse>>any());
+    int opCount = 0;
+    ApiFuture<WriteResult> result1 = bulkWriter.set(doc1, LocalFirestoreHelper.SINGLE_FIELD_MAP);
+    ApiFuture<WriteResult> result2 = bulkWriter.set(doc2, LocalFirestoreHelper.SINGLE_FIELD_MAP);
+    bulkWriter.close();
+
+    for (ApiFuture<WriteResult> result : Arrays.asList(result1, result2)) {
+      try {
+        result.get();
+      } catch (Exception e) {
+        assertTrue(
+            e.getMessage()
+                .contains("java.lang.IllegalStateException: Mock batchWrite failed in test"));
+        ++opCount;
+      }
+    }
+    assertEquals(2, opCount);
+  }
+
+  @Test
+  public void retriesWritesWhenBatchWriteFailsWithRetryableError() throws Exception {
+    FirestoreException retryableError =
+        FirestoreException.serverRejected(
+            Status.fromCode(Status.Code.ABORTED), "Mock batchWrite failed in test");
+
+    ApiFuture<Void> errorFuture = ApiFutures.immediateFailedFuture(retryableError);
+
+    doReturn(errorFuture)
+        .doReturn(errorFuture)
+        .doReturn(errorFuture)
+        .doReturn(successResponse(3))
+        .when(firestoreMock)
+        .sendRequest(
+            batchWriteCapture.capture(),
+            Matchers.<UnaryCallable<BatchWriteRequest, BatchWriteResponse>>any());
+
+    ApiFuture<WriteResult> result = bulkWriter.set(doc1, LocalFirestoreHelper.SINGLE_FIELD_MAP);
+    bulkWriter.close();
+
+    assertEquals(Timestamp.ofTimeSecondsAndNanos(3, 0), result.get().getUpdateTime());
+  }
+
+  @Test
+  public void failsWritesAfterAllRetryAttemptsFail() throws Exception {
+    final int[] retryAttempts = {0};
+    doAnswer(
+            new Answer() {
+              public ApiFuture<Object> answer(InvocationOnMock mock) {
+                retryAttempts[0]++;
+                return ApiFutures.immediateFailedFuture(
+                    FirestoreException.serverRejected(
+                        Status.fromCode(Status.Code.ABORTED), "Mock batchWrite failed in test"));
+              }
+            })
+        .when(firestoreMock)
+        .sendRequest(
+            batchWriteCapture.capture(),
+            Matchers.<UnaryCallable<BatchWriteRequest, BatchWriteResponse>>any());
+    ApiFuture<WriteResult> result = bulkWriter.set(doc1, LocalFirestoreHelper.SINGLE_FIELD_MAP);
+    bulkWriter.close();
+
+    try {
+      result.get();
+      Assert.fail("Expected set() operation to fail");
+    } catch (Exception e) {
+      assertTrue(e.getMessage().contains("Mock batchWrite failed in test"));
+      assertEquals(retryAttempts[0], BulkWriter.MAX_RETRY_ATTEMPTS + 1);
+    }
+  }
+}
diff --git a/google-cloud-firestore/src/test/java/com/google/cloud/firestore/LocalFirestoreHelper.java b/google-cloud-firestore/src/test/java/com/google/cloud/firestore/LocalFirestoreHelper.java
index a57f36c05..db022ba95 100644
--- a/google-cloud-firestore/src/test/java/com/google/cloud/firestore/LocalFirestoreHelper.java
+++ b/google-cloud-firestore/src/test/java/com/google/cloud/firestore/LocalFirestoreHelper.java
@@ -17,19 +17,25 @@
 package com.google.cloud.firestore;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+import static org.mockito.Mockito.doAnswer;
 
 import com.fasterxml.jackson.core.type.TypeReference;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.api.core.ApiFuture;
 import com.google.api.core.ApiFutures;
+import com.google.api.core.SettableApiFuture;
 import com.google.api.gax.retrying.RetrySettings;
 import com.google.api.gax.rpc.ApiStreamObserver;
+import com.google.api.gax.rpc.UnaryCallable;
 import com.google.cloud.Timestamp;
+import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.firestore.v1.ArrayValue;
 import com.google.firestore.v1.BatchGetDocumentsRequest;
 import com.google.firestore.v1.BatchGetDocumentsResponse;
+import com.google.firestore.v1.BatchWriteRequest;
 import com.google.firestore.v1.BeginTransactionRequest;
 import com.google.firestore.v1.BeginTransactionResponse;
 import com.google.firestore.v1.CommitRequest;
@@ -53,6 +59,8 @@
 import com.google.firestore.v1.Write;
 import com.google.protobuf.ByteString;
 import com.google.protobuf.Empty;
+import com.google.protobuf.GeneratedMessageV3;
+import com.google.protobuf.Message;
 import com.google.protobuf.NullValue;
 import com.google.type.LatLng;
 import java.io.IOException;
@@ -65,13 +73,18 @@
 import java.util.Comparator;
 import java.util.Date;
 import java.util.HashMap;
+import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.concurrent.Semaphore;
 import java.util.concurrent.TimeUnit;
 import javax.annotation.Nullable;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Matchers;
 import org.mockito.invocation.InvocationOnMock;
 import org.mockito.stubbing.Answer;
+import org.mockito.stubbing.Stubber;
 import org.threeten.bp.Duration;
 
 public final class LocalFirestoreHelper {
@@ -88,6 +101,7 @@ public final class LocalFirestoreHelper {
   public static final String COLLECTION_ID;
   public static final String DOCUMENT_PATH;
   public static final String DOCUMENT_NAME;
+  public static final String DOCUMENT_ROOT;
   public static final String TRANSACTION_ID;
 
   public static final Map<String, Value> EMPTY_MAP_PROTO;
@@ -365,23 +379,36 @@ public static List<FieldTransform> transform(
     return transforms;
   }
 
-  public static Write create(Map<String, Value> fields) {
+  public static Write create(Map<String, Value> fields, String docPath) {
     Write.Builder write = Write.newBuilder();
     Document.Builder document = write.getUpdateBuilder();
-    document.setName(DOCUMENT_NAME);
+    document.setName(DOCUMENT_ROOT + docPath);
     document.putAllFields(fields);
     write.getCurrentDocumentBuilder().setExists(false);
     return write.build();
   }
 
+  public static Write create(Map<String, Value> fields) {
+    return create(fields, DOCUMENT_PATH);
+  }
+
   public static Write set(Map<String, Value> fields) {
-    return set(fields, null);
+    return set(fields, null, DOCUMENT_PATH);
+  }
+
+  public static Write set(Map<String, Value> fields, String docPath) {
+    return set(fields, null, docPath);
   }
 
   public static Write set(Map<String, Value> fields, @Nullable List<String> fieldMap) {
+    return set(fields, fieldMap, DOCUMENT_PATH);
+  }
+
+  public static Write set(
+      Map<String, Value> fields, @Nullable List<String> fieldMap, String docPath) {
     Write.Builder write = Write.newBuilder();
     Document.Builder document = write.getUpdateBuilder();
-    document.setName(DOCUMENT_NAME);
+    document.setName(DOCUMENT_ROOT + docPath);
     document.putAllFields(fields);
 
     if (fieldMap != null) {
@@ -392,9 +419,7 @@ public static Write set(Map<String, Value> fields, @Nullable List<String> fieldM
   }
 
   public static Write delete() {
-    Write.Builder write = Write.newBuilder();
-    write.setDelete(DOCUMENT_NAME);
-    return write.build();
+    return delete(DOCUMENT_PATH);
   }
 
   public static Write delete(Precondition precondition) {
@@ -404,24 +429,38 @@ public static Write delete(Precondition precondition) {
     return write.build();
   }
 
-  public static Write update(Map<String, Value> fields, List<String> fieldMap) {
+  public static Write delete(String docPath) {
     Write.Builder write = Write.newBuilder();
-    Document.Builder document = write.getUpdateBuilder();
-    document.setName(DOCUMENT_NAME);
-    document.putAllFields(fields);
-    write.getUpdateMaskBuilder().addAllFieldPaths(fieldMap);
-    write.getCurrentDocumentBuilder().setExists(true);
+    write.setDelete(DOCUMENT_ROOT + docPath);
     return write.build();
   }
 
+  public static Write update(Map<String, Value> fields, List<String> fieldMap) {
+    return update(fields, fieldMap, UPDATE_PRECONDITION);
+  }
+
+  public static Write update(
+      Map<String, Value> fields, List<String> fieldMap, @Nullable Precondition precondition) {
+    return update(fields, fieldMap, precondition, DOCUMENT_PATH);
+  }
+
+  public static Write update(Map<String, Value> fields, String docPath) {
+    return update(fields, Collections.singletonList("foo"), UPDATE_PRECONDITION, docPath);
+  }
+
   public static Write update(
-      Map<String, Value> fields, @Nullable List<String> fieldMap, Precondition precondition) {
+      Map<String, Value> fields,
+      @Nullable List<String> fieldMap,
+      @Nullable Precondition precondition,
+      String docPath) {
     Write.Builder write = Write.newBuilder();
     Document.Builder document = write.getUpdateBuilder();
-    document.setName(DOCUMENT_NAME);
+    document.setName(DOCUMENT_ROOT + docPath);
     document.putAllFields(fields);
     write.getUpdateMaskBuilder().addAllFieldPaths(fieldMap);
-    write.getCurrentDocumentBuilder().mergeFrom(precondition);
+    if (precondition != null) {
+      write.getCurrentDocumentBuilder().mergeFrom(precondition);
+    }
     return write.build();
   }
 
@@ -445,6 +484,13 @@ public static CommitRequest commit(Write write, List<FieldTransform> transforms)
     return commit((String) null, write.toBuilder().addAllUpdateTransforms(transforms).build());
   }
 
+  public static BatchWriteRequest batchWrite(Write... writes) {
+    BatchWriteRequest.Builder batchWriteRequest = BatchWriteRequest.newBuilder();
+    batchWriteRequest.setDatabase(DATABASE_NAME);
+    batchWriteRequest.addAllWrites(Arrays.asList(writes));
+    return batchWriteRequest.build();
+  }
+
   public static StructuredQuery filter(StructuredQuery.FieldFilter.Operator operator) {
     return filter(operator, "foo", "bar");
   }
@@ -719,6 +765,7 @@ public boolean equals(Object o) {
     COLLECTION_ID = "coll";
     DOCUMENT_PATH = "coll/doc";
     DOCUMENT_NAME = DATABASE_NAME + "/documents/" + DOCUMENT_PATH;
+    DOCUMENT_ROOT = DATABASE_NAME + "/documents/";
 
     EMPTY_MAP_PROTO =
         map("inner", Value.newBuilder().setMapValue(MapValue.getDefaultInstance()).build());
@@ -883,4 +930,83 @@ private static Map<String, Object> fromJsonString(String json) {
   public static Map<String, Object> fromSingleQuotedString(String json) {
     return fromJsonString(json.replace("'", "\""));
   }
+
+  /**
+   * Contains a map of request/response pairs that are used to create stub responses when
+   * `sendRequest()` is called.
+   */
+  static class ResponseStubber
+      extends LinkedHashMap<GeneratedMessageV3, ApiFuture<? extends GeneratedMessageV3>> {
+
+    /**
+     * Verifies the response before returning. This method can be overridden to perform logic before
+     * the stubbed response is returned.
+     */
+    ApiFuture<? extends GeneratedMessageV3> verifyResponse(
+        ApiFuture<? extends GeneratedMessageV3> response) {
+      return response;
+    }
+
+    void initializeStub(
+        ArgumentCaptor<? extends Message> argumentCaptor, FirestoreImpl firestoreMock) {
+      Stubber stubber = null;
+      for (final ApiFuture<? extends GeneratedMessageV3> response : values()) {
+        Answer<ApiFuture<? extends GeneratedMessageV3>> answer =
+            new Answer<ApiFuture<? extends GeneratedMessageV3>>() {
+              @Override
+              public ApiFuture<? extends GeneratedMessageV3> answer(
+                  InvocationOnMock invocationOnMock) throws Throwable {
+                return verifyResponse(response);
+              }
+            };
+        stubber = (stubber != null) ? stubber.doAnswer(answer) : doAnswer(answer);
+      }
+      Preconditions.checkNotNull(stubber, "Stubber should not be null");
+      stubber
+          .when(firestoreMock)
+          .sendRequest(argumentCaptor.capture(), Matchers.<UnaryCallable<Message, Message>>any());
+    }
+  }
+
+  /**
+   * Contains a map of request/response pairs that are used to create stub responses when
+   * `sendRequest()` is called.
+   *
+   * <p>Enforces that only one active request can be pending at a time.
+   */
+  static class SerialResponseStubber extends ResponseStubber {
+    int activeRequestCounter = 0;
+    SettableApiFuture<Void> activeRequestComplete = SettableApiFuture.create();
+    Semaphore semaphore = new Semaphore(0);
+
+    void markAllRequestsComplete() {
+      activeRequestComplete.set(null);
+    }
+
+    void awaitRequest() {
+      activeRequestComplete = SettableApiFuture.create();
+      try {
+        semaphore.acquire();
+      } catch (Exception e) {
+        fail("sempahore.acquire() should not fail");
+      }
+    }
+
+    @Override
+    ApiFuture<? extends GeneratedMessageV3> verifyResponse(
+        ApiFuture<? extends GeneratedMessageV3> response) {
+      ++activeRequestCounter;
+
+      // This assert is used to test that only one request is made at a time.
+      assertEquals(1, activeRequestCounter);
+      try {
+        semaphore.release();
+        activeRequestComplete.get();
+      } catch (Exception e) {
+        fail("activeRequestComplete.get() should not fail");
+      }
+      --activeRequestCounter;
+      return response;
+    }
+  }
 }
diff --git a/google-cloud-firestore/src/test/java/com/google/cloud/firestore/TransactionTest.java b/google-cloud-firestore/src/test/java/com/google/cloud/firestore/TransactionTest.java
index f7b3c15dc..ac3767c76 100644
--- a/google-cloud-firestore/src/test/java/com/google/cloud/firestore/TransactionTest.java
+++ b/google-cloud-firestore/src/test/java/com/google/cloud/firestore/TransactionTest.java
@@ -50,6 +50,7 @@
 import com.google.api.gax.rpc.StatusCode;
 import com.google.api.gax.rpc.UnaryCallable;
 import com.google.cloud.Timestamp;
+import com.google.cloud.firestore.LocalFirestoreHelper.ResponseStubber;
 import com.google.cloud.firestore.spi.v1.FirestoreRpc;
 import com.google.firestore.v1.BatchGetDocumentsRequest;
 import com.google.firestore.v1.DocumentMask;
@@ -60,7 +61,6 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
-import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ExecutionException;
@@ -77,7 +77,6 @@
 import org.mockito.Mockito;
 import org.mockito.Spy;
 import org.mockito.runners.MockitoJUnitRunner;
-import org.mockito.stubbing.Stubber;
 
 @RunWith(MockitoJUnitRunner.class)
 public class TransactionTest {
@@ -338,8 +337,8 @@ public ApiFuture<String> updateCallback(Transaction transaction) {
 
   @Test
   public void limitsRetriesWithFailure() {
-    RequestResponseMap requestResponseMap =
-        new RequestResponseMap() {
+    ResponseStubber responseStubber =
+        new ResponseStubber() {
           {
             put(begin(), beginResponse("foo1"));
             put(commit("foo1"), RETRYABLE_API_EXCEPTION);
@@ -359,7 +358,7 @@ public void limitsRetriesWithFailure() {
           }
         };
 
-    initializeStub(requestResponseMap);
+    responseStubber.initializeStub(requestCapture, firestoreMock);
 
     final AtomicInteger retryCount = new AtomicInteger(1);
 
@@ -381,18 +380,18 @@ public String updateCallback(Transaction transaction) {
     }
 
     List<Message> requests = requestCapture.getAllValues();
-    assertEquals(requestResponseMap.size(), requests.size());
+    assertEquals(responseStubber.size(), requests.size());
 
     int index = 0;
-    for (GeneratedMessageV3 request : requestResponseMap.keySet()) {
+    for (GeneratedMessageV3 request : responseStubber.keySet()) {
       assertEquals(request, requests.get(index++));
     }
   }
 
   @Test
   public void limitsRetriesWithSuccess() throws Exception {
-    RequestResponseMap requestResponseMap =
-        new RequestResponseMap() {
+    ResponseStubber responseStubber =
+        new ResponseStubber() {
           {
             put(begin(), beginResponse("foo1"));
             put(commit("foo1"), RETRYABLE_API_EXCEPTION);
@@ -414,7 +413,7 @@ public void limitsRetriesWithSuccess() throws Exception {
           }
         };
 
-    initializeStub(requestResponseMap);
+    responseStubber.initializeStub(requestCapture, firestoreMock);
 
     final AtomicInteger retryCount = new AtomicInteger(1);
 
@@ -431,10 +430,10 @@ public String updateCallback(Transaction transaction) {
     assertEquals("foo6", transaction.get());
 
     List<Message> requests = requestCapture.getAllValues();
-    assertEquals(requestResponseMap.size(), requests.size());
+    assertEquals(responseStubber.size(), requests.size());
 
     int index = 0;
-    for (GeneratedMessageV3 request : requestResponseMap.keySet()) {
+    for (GeneratedMessageV3 request : responseStubber.keySet()) {
       assertEquals(request, requests.get(index++));
     }
   }
@@ -470,8 +469,8 @@ public void retriesBasedOnErrorCode() throws Exception {
           new ApiException(new Exception("Test Exception"), code, shouldRetry);
 
       if (shouldRetry) {
-        RequestResponseMap requestResponseMap =
-            new RequestResponseMap() {
+        ResponseStubber responseStubber =
+            new ResponseStubber() {
               {
                 put(begin(), beginResponse("foo1"));
                 put(
@@ -482,7 +481,7 @@ public void retriesBasedOnErrorCode() throws Exception {
                 put(commit("foo2"), commitResponse(0, 0));
               }
             };
-        initializeStub(requestResponseMap);
+        responseStubber.initializeStub(requestCapture, firestoreMock);
 
         final int[] attempts = new int[] {0};
 
@@ -500,8 +499,8 @@ public String updateCallback(Transaction transaction) {
 
         assertEquals(2, attempts[0]);
       } else {
-        RequestResponseMap requestResponseMap =
-            new RequestResponseMap() {
+        ResponseStubber responseStubber =
+            new ResponseStubber() {
               {
                 put(begin(), beginResponse("foo1"));
                 put(
@@ -511,7 +510,7 @@ public String updateCallback(Transaction transaction) {
               }
             };
 
-        initializeStub(requestResponseMap);
+        responseStubber.initializeStub(requestCapture, firestoreMock);
 
         final int[] attempts = new int[] {0};
 
@@ -866,17 +865,4 @@ public String updateCallback(Transaction transaction) {
     assertEquals(begin(), requests.get(0));
     assertEquals(commit(TRANSACTION_ID, writes.toArray(new Write[] {})), requests.get(1));
   }
-
-  static class RequestResponseMap
-      extends LinkedHashMap<GeneratedMessageV3, ApiFuture<? extends GeneratedMessageV3>> {}
-
-  private void initializeStub(RequestResponseMap requestResponseMap) {
-    Stubber stubber = null;
-    for (ApiFuture<? extends GeneratedMessageV3> response : requestResponseMap.values()) {
-      stubber = (stubber != null) ? stubber.doReturn(response) : doReturn(response);
-    }
-    stubber
-        .when(firestoreMock)
-        .sendRequest(requestCapture.capture(), Matchers.<UnaryCallable<Message, Message>>any());
-  }
 }
diff --git a/google-cloud-firestore/src/test/java/com/google/cloud/firestore/WriteBatchTest.java b/google-cloud-firestore/src/test/java/com/google/cloud/firestore/WriteBatchTest.java
index 3e629324f..ecf2930c3 100644
--- a/google-cloud-firestore/src/test/java/com/google/cloud/firestore/WriteBatchTest.java
+++ b/google-cloud-firestore/src/test/java/com/google/cloud/firestore/WriteBatchTest.java
@@ -18,6 +18,7 @@
 
 import static com.google.cloud.firestore.LocalFirestoreHelper.UPDATED_SINGLE_FIELD_PROTO;
 import static com.google.cloud.firestore.LocalFirestoreHelper.UPDATE_SINGLE_FIELD_OBJECT;
+import static com.google.cloud.firestore.LocalFirestoreHelper.batchWrite;
 import static com.google.cloud.firestore.LocalFirestoreHelper.commit;
 import static com.google.cloud.firestore.LocalFirestoreHelper.commitResponse;
 import static com.google.cloud.firestore.LocalFirestoreHelper.create;
@@ -26,14 +27,19 @@
 import static com.google.cloud.firestore.LocalFirestoreHelper.set;
 import static com.google.cloud.firestore.LocalFirestoreHelper.update;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
 import static org.mockito.Mockito.doReturn;
 
+import com.google.api.core.ApiFutures;
 import com.google.api.gax.rpc.UnaryCallable;
 import com.google.cloud.Timestamp;
 import com.google.cloud.firestore.spi.v1.FirestoreRpc;
+import com.google.firestore.v1.BatchWriteRequest;
+import com.google.firestore.v1.BatchWriteResponse;
 import com.google.firestore.v1.CommitRequest;
 import com.google.firestore.v1.CommitResponse;
 import com.google.firestore.v1.Write;
+import io.grpc.Status;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
@@ -52,12 +58,13 @@
 public class WriteBatchTest {
 
   @Spy
-  private FirestoreImpl firestoreMock =
+  private final FirestoreImpl firestoreMock =
       new FirestoreImpl(
           FirestoreOptions.newBuilder().setProjectId("test-project").build(),
           Mockito.mock(FirestoreRpc.class));
 
   @Captor private ArgumentCaptor<CommitRequest> commitCapture;
+  @Captor private ArgumentCaptor<BatchWriteRequest> batchWriteCapture;
 
   private WriteBatch batch;
   private DocumentReference documentReference;
@@ -329,4 +336,40 @@ public void deleteDocument() throws Exception {
     CommitRequest commitRequest = commitCapture.getValue();
     assertEquals(commit(writes.toArray(new Write[] {})), commitRequest);
   }
+
+  @Test
+  public void bulkCommit() throws Exception {
+    BatchWriteResponse.Builder response = BatchWriteResponse.newBuilder();
+    response.addWriteResultsBuilder().getUpdateTimeBuilder().setNanos(1);
+    response.addWriteResultsBuilder();
+    response.addStatusBuilder().build();
+    response.addStatusBuilder().setCode(14).build();
+    doReturn(ApiFutures.immediateFuture(response.build()))
+        .when(firestoreMock)
+        .sendRequest(
+            batchWriteCapture.capture(),
+            Matchers.<UnaryCallable<BatchWriteRequest, BatchWriteResponse>>any());
+
+    List<Write> writes = new ArrayList<>();
+    batch.set(documentReference, LocalFirestoreHelper.SINGLE_FIELD_MAP);
+    writes.add(set(LocalFirestoreHelper.SINGLE_FIELD_PROTO));
+
+    batch.create(documentReference, LocalFirestoreHelper.SINGLE_FIELD_MAP);
+    writes.add(create(LocalFirestoreHelper.SINGLE_FIELD_PROTO));
+
+    assertEquals(2, batch.getMutationsSize());
+
+    batch.markReadyToSend();
+    List<BatchWriteResult> batchWriteResults = batch.bulkCommit().get();
+
+    assertNull(batchWriteResults.get(0).getException());
+    assertEquals(Timestamp.ofTimeSecondsAndNanos(0, 1), batchWriteResults.get(0).getWriteTime());
+    assertEquals(
+        Status.UNAVAILABLE,
+        ((FirestoreException) batchWriteResults.get(1).getException()).getStatus());
+    assertNull(batchWriteResults.get(1).getWriteTime());
+
+    BatchWriteRequest batchWriteRequest = batchWriteCapture.getValue();
+    assertEquals(batchWrite(writes.toArray(new Write[] {})), batchWriteRequest);
+  }
 }
diff --git a/google-cloud-firestore/src/test/java/com/google/cloud/firestore/it/ITSystemTest.java b/google-cloud-firestore/src/test/java/com/google/cloud/firestore/it/ITSystemTest.java
index c41855de9..d2a768181 100644
--- a/google-cloud-firestore/src/test/java/com/google/cloud/firestore/it/ITSystemTest.java
+++ b/google-cloud-firestore/src/test/java/com/google/cloud/firestore/it/ITSystemTest.java
@@ -33,6 +33,7 @@
 import com.google.api.core.SettableApiFuture;
 import com.google.api.gax.rpc.ApiStreamObserver;
 import com.google.cloud.Timestamp;
+import com.google.cloud.firestore.BulkWriter;
 import com.google.cloud.firestore.CollectionReference;
 import com.google.cloud.firestore.DocumentReference;
 import com.google.cloud.firestore.DocumentSnapshot;
@@ -1306,6 +1307,113 @@ public void deleteNestedFieldUsingFieldPath() throws Exception {
     assertNull(documentSnapshots.getData().get("c.d"));
   }
 
+  @Test
+  public void bulkWriterCreate() throws Exception {
+    DocumentReference docRef = randomColl.document();
+
+    BulkWriter writer = firestore.bulkWriter();
+    ApiFuture<WriteResult> result =
+        writer.create(docRef, Collections.singletonMap("foo", (Object) "bar"));
+    writer.close();
+
+    assertNotNull(result.get().getUpdateTime());
+    DocumentSnapshot snapshot = docRef.get().get();
+    assertEquals("bar", snapshot.get("foo"));
+  }
+
+  @Test
+  public void bulkWriterCreateAddsPrecondition() throws Exception {
+    DocumentReference docRef = randomColl.document();
+    docRef.set(Collections.singletonMap("foo", (Object) "bar")).get();
+
+    BulkWriter writer = firestore.bulkWriter();
+    ApiFuture<WriteResult> result =
+        writer.create(docRef, Collections.singletonMap("foo", (Object) "bar"));
+    writer.close();
+
+    try {
+      result.get();
+      fail("Create operation should have thrown exception");
+    } catch (Exception e) {
+      assertTrue(e.getMessage().contains("Document already exists"));
+    }
+  }
+
+  @Test
+  public void bulkWriterSet() throws Exception {
+    DocumentReference docRef = randomColl.document();
+
+    BulkWriter writer = firestore.bulkWriter();
+    ApiFuture<WriteResult> result =
+        writer.set(docRef, Collections.singletonMap("foo", (Object) "bar"));
+    writer.close();
+
+    assertNotNull(result.get().getUpdateTime());
+    DocumentSnapshot snapshot = docRef.get().get();
+    assertEquals("bar", snapshot.get("foo"));
+  }
+
+  @Test
+  public void bulkWriterUpdate() throws Exception {
+    DocumentReference docRef = randomColl.document();
+    docRef.set(Collections.singletonMap("foo", "oldValue")).get();
+
+    BulkWriter writer = firestore.bulkWriter();
+    ApiFuture<WriteResult> result = writer.update(docRef, "foo", "newValue");
+    writer.close();
+
+    assertNotNull(result.get().getUpdateTime());
+    DocumentSnapshot snapshot = docRef.get().get();
+    assertEquals("newValue", snapshot.get("foo"));
+  }
+
+  @Test
+  public void bulkWriterUpdateAddsPrecondition() throws Exception {
+    DocumentReference docRef = randomColl.document();
+
+    BulkWriter writer = firestore.bulkWriter();
+    ApiFuture<WriteResult> result = writer.update(docRef, "foo", "newValue");
+    writer.close();
+
+    try {
+      result.get();
+      fail("Update operation should have thrown exception");
+    } catch (Exception e) {
+      assertTrue(e.getMessage().contains("No document to update"));
+    }
+  }
+
+  @Test
+  public void bulkWriterDelete() throws Exception {
+    DocumentReference docRef = randomColl.document();
+    docRef.set(Collections.singletonMap("foo", "oldValue")).get();
+
+    BulkWriter writer = firestore.bulkWriter();
+    ApiFuture<WriteResult> result = writer.delete(docRef);
+    writer.close();
+
+    assertNotNull(result.get().getUpdateTime());
+    // TODO(b/158502664): Remove this check once we can get write times.
+    assertEquals(Timestamp.ofTimeSecondsAndNanos(0, 0), result.get().getUpdateTime());
+    DocumentSnapshot snapshot = docRef.get().get();
+    assertNull(snapshot.get("foo"));
+  }
+
+  @Test
+  public void bulkWriterWritesInOrder() throws Exception {
+    DocumentReference docRef = randomColl.document();
+    docRef.set(Collections.singletonMap("foo", "oldValue")).get();
+
+    BulkWriter writer = firestore.bulkWriter();
+    writer.set(docRef, Collections.singletonMap("foo", (Object) "bar1"));
+    writer.set(docRef, Collections.singletonMap("foo", (Object) "bar2"));
+    writer.set(docRef, Collections.singletonMap("foo", (Object) "bar3"));
+    writer.close();
+
+    ApiFuture<DocumentSnapshot> result = docRef.get();
+    assertEquals(Collections.singletonMap("foo", "bar3"), result.get().getData());
+  }
+
   /** Wrapper around ApiStreamObserver that returns the results in a list. */
   private static class StreamConsumer<T> implements ApiStreamObserver<T> {
     SettableApiFuture<List<T>> done = SettableApiFuture.create();
diff --git a/pom.xml b/pom.xml
index df6a385f0..8aa4ed677 100644
--- a/pom.xml
+++ b/pom.xml
@@ -177,6 +177,13 @@
         <version>1.35.3-SNAPSHOT</version><!-- {x-version-update:grpc-google-cloud-firestore-v1:current} -->
       </dependency>
 
+      <!-- Remove after google-cloud-shared-dependencies is up to v0.8.4 -->
+      <dependency>
+        <groupId>com.google.api</groupId>
+        <artifactId>api-common</artifactId>
+        <version>1.10.0</version>
+      </dependency>
+
       <dependency>
         <groupId>com.google.cloud</groupId>
         <artifactId>google-cloud-shared-dependencies</artifactId>