Repository: googleapis/java-storage
Pull request #1045 (2021-09-24 22:32:00)

chore(deps): update dependency com.google.cloud:google-cloud-storage to v2.1.6
[![WhiteSource Renovate](https://app.renovatebot.com/images/banner.svg)](https://renovatebot.com)

This PR contains the following updates:

| Package | Change | Age | Adoption | Passing | Confidence |
|---|---|---|---|---|---|
| [com.google.cloud:google-cloud-storage](https://togithub.com/googleapis/java-storage) | `2.1.5` -> `2.1.6` | [![age](https://badges.renovateapi.com/packages/maven/com.google.cloud:google-cloud-storage/2.1.6/age-slim)](https://docs.renovatebot.com/merge-confidence/) | [![adoption](https://badges.renovateapi.com/packages/maven/com.google.cloud:google-cloud-storage/2.1.6/adoption-slim)](https://docs.renovatebot.com/merge-confidence/) | [![passing](https://badges.renovateapi.com/packages/maven/com.google.cloud:google-cloud-storage/2.1.6/compatibility-slim/2.1.5)](https://docs.renovatebot.com/merge-confidence/) | [![confidence](https://badges.renovateapi.com/packages/maven/com.google.cloud:google-cloud-storage/2.1.6/confidence-slim/2.1.5)](https://docs.renovatebot.com/merge-confidence/) |

---

### Release Notes

<details>
<summary>googleapis/java-storage</summary>

### [`v2.1.6`](https://togithub.com/googleapis/java-storage/blob/master/CHANGELOG.md#&#8203;216-httpswwwgithubcomgoogleapisjava-storagecomparev215v216-2021-09-24)

[Compare Source](https://togithub.com/googleapis/java-storage/compare/v2.1.5...v2.1.6)

</details>

---

### Configuration

📅 **Schedule**: At any time (no schedule defined).

🚦 **Automerge**: Disabled by config. Please merge this manually once you are satisfied.

♻ **Rebasing**: Whenever PR becomes conflicted, or you tick the rebase/retry checkbox.

🔕 **Ignore**: Close this PR and you won't be reminded about this update again.

---

 - [ ] <!-- rebase-check -->If you want to rebase/retry this PR, check this box.

---

This PR has been generated by [WhiteSource Renovate](https://renovate.whitesourcesoftware.com). View repository job log [here](https://app.renovatebot.com/dashboard#github/googleapis/java-storage).


Patch:
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/Blob.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/Blob.java
index abc61eb1..4fbbe069 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/Blob.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/Blob.java
@@ -16,7 +16,6 @@
 
 package com.google.cloud.storage;
 
-import static com.google.cloud.RetryHelper.runWithRetries;
 import static com.google.cloud.storage.Blob.BlobSourceOption.toGetOptions;
 import static com.google.cloud.storage.Blob.BlobSourceOption.toSourceOptions;
 import static com.google.common.base.Preconditions.checkNotNull;
@@ -25,9 +24,8 @@ import static java.util.concurrent.Executors.callable;
 import com.google.api.services.storage.model.StorageObject;
 import com.google.auth.ServiceAccountSigner;
 import com.google.auth.ServiceAccountSigner.SigningException;
+import com.google.cloud.ExceptionHandler;
 import com.google.cloud.ReadChannel;
-import com.google.cloud.RetryHelper;
-import com.google.cloud.Tuple;
 import com.google.cloud.WriteChannel;
 import com.google.cloud.storage.Acl.Entity;
 import com.google.cloud.storage.Storage.BlobTargetOption;
@@ -35,7 +33,6 @@ import com.google.cloud.storage.Storage.BlobWriteOption;
 import com.google.cloud.storage.Storage.CopyRequest;
 import com.google.cloud.storage.Storage.SignUrlOption;
 import com.google.cloud.storage.spi.v1.StorageRpc;
-import com.google.common.base.Function;
 import com.google.common.io.BaseEncoding;
 import com.google.common.io.CountingOutputStream;
 import java.io.IOException;
@@ -50,6 +47,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
 
 /**
  * An object in Google Cloud Storage. A {@code Blob} object includes the {@code BlobId} instance,
@@ -78,18 +76,9 @@ public class Blob extends BlobInfo {
   private static final long serialVersionUID = -6806832496717441434L;
 
   private final StorageOptions options;
+  private final RetryAlgorithmManager retryAlgorithmManager;
   private transient Storage storage;
 
-  static final Function<Tuple<Storage, StorageObject>, Blob> BLOB_FROM_PB_FUNCTION =
-      new Function<Tuple<Storage, StorageObject>, Blob>() {
-        @Override
-        public Blob apply(Tuple<Storage, StorageObject> pb) {
-          return Blob.fromPb(pb.x(), pb.y());
-        }
-      };
-
-  private static final int DEFAULT_CHUNK_SIZE = 2 * 1024 * 1024;
-
   /** Class for specifying blob source options when {@code Blob} methods are used. */
   public static class BlobSourceOption extends Option {
 
@@ -258,26 +247,18 @@ public class Blob extends BlobInfo {
   public void downloadTo(OutputStream outputStream, BlobSourceOption... options) {
     final CountingOutputStream countingOutputStream = new CountingOutputStream(outputStream);
     final StorageRpc storageRpc = this.options.getStorageRpcV1();
+    StorageObject pb = getBlobId().toPb();
     final Map<StorageRpc.Option, ?> requestOptions = StorageImpl.optionMap(getBlobId(), options);
-    try {
-      runWithRetries(
-          callable(
-              new Runnable() {
-                @Override
-                public void run() {
-                  storageRpc.read(
-                      getBlobId().toPb(),
-                      requestOptions,
-                      countingOutputStream.getCount(),
-                      countingOutputStream);
-                }
-              }),
-          this.options.getRetrySettings(),
-          StorageImpl.EXCEPTION_HANDLER,
-          this.options.getClock());
-    } catch (RetryHelper.RetryHelperException e) {
-      StorageException.translateAndThrow(e);
-    }
+    ExceptionHandler exceptionHandler = retryAlgorithmManager.getForObjectsGet(pb, requestOptions);
+    Retrying.run(
+        this.options,
+        exceptionHandler,
+        callable(
+            () -> {
+              storageRpc.read(
+                  pb, requestOptions, countingOutputStream.getCount(), countingOutputStream);
+            }),
+        Function.identity());
   }
 
   /**
@@ -506,6 +487,7 @@ public class Blob extends BlobInfo {
     super(infoBuilder);
     this.storage = checkNotNull(storage);
     this.options = storage.getOptions();
+    this.retryAlgorithmManager = storage.getOptions().getRetryAlgorithmManager();
   }
 
   /**
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobReadChannel.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobReadChannel.java
index c5fce5bd..59d6188d 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobReadChannel.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobReadChannel.java
@@ -19,6 +19,7 @@ package com.google.cloud.storage;
 import static com.google.cloud.RetryHelper.runWithRetries;
 
 import com.google.api.services.storage.model.StorageObject;
+import com.google.cloud.ExceptionHandler;
 import com.google.cloud.ReadChannel;
 import com.google.cloud.RestorableState;
 import com.google.cloud.RetryHelper;
@@ -31,7 +32,6 @@ import java.nio.ByteBuffer;
 import java.nio.channels.ClosedChannelException;
 import java.util.Map;
 import java.util.Objects;
-import java.util.concurrent.Callable;
 
 /** Default implementation for ReadChannel. */
 class BlobReadChannel implements ReadChannel {
@@ -41,6 +41,7 @@ class BlobReadChannel implements ReadChannel {
   private final StorageOptions serviceOptions;
   private final BlobId blob;
   private final Map<StorageRpc.Option, ?> requestOptions;
+  private final RetryAlgorithmManager retryAlgorithmManager;
   private String lastEtag;
   private long position;
   private boolean isOpen;
@@ -57,6 +58,7 @@ class BlobReadChannel implements ReadChannel {
     this.serviceOptions = serviceOptions;
     this.blob = blob;
     this.requestOptions = requestOptions;
+    this.retryAlgorithmManager = serviceOptions.getRetryAlgorithmManager();
     isOpen = true;
     storageRpc = serviceOptions.getStorageRpcV1();
     storageObject = blob.toPb();
@@ -119,24 +121,21 @@ class BlobReadChannel implements ReadChannel {
       }
       final int toRead = Math.max(byteBuffer.remaining(), chunkSize);
       try {
+        ExceptionHandler exceptionHandler =
+            retryAlgorithmManager.getForObjectsGet(storageObject, requestOptions);
         Tuple<String, byte[]> result =
             runWithRetries(
-                new Callable<Tuple<String, byte[]>>() {
-                  @Override
-                  public Tuple<String, byte[]> call() {
-                    return storageRpc.read(storageObject, requestOptions, position, toRead);
-                  }
-                },
+                () -> storageRpc.read(storageObject, requestOptions, position, toRead),
                 serviceOptions.getRetrySettings(),
-                StorageImpl.EXCEPTION_HANDLER,
+                exceptionHandler,
                 serviceOptions.getClock());
-        if (result.y().length > 0 && lastEtag != null && !Objects.equals(result.x(), lastEtag)) {
-          StringBuilder messageBuilder = new StringBuilder();
-          messageBuilder.append("Blob ").append(blob).append(" was updated while reading");
-          throw new IOException(messageBuilder.toString());
+        String etag = result.x();
+        byte[] bytes = result.y();
+        if (bytes.length > 0 && lastEtag != null && !Objects.equals(etag, lastEtag)) {
+          throw new IOException("Blob " + blob + " was updated while reading");
         }
-        lastEtag = result.x();
-        buffer = result.y();
+        lastEtag = etag;
+        buffer = bytes;
       } catch (RetryHelper.RetryHelperException e) {
         throw new IOException(e);
       }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteChannel.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteChannel.java
index 6207ef66..fd7e73e3 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteChannel.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteChannel.java
@@ -17,47 +17,41 @@
 package com.google.cloud.storage;
 
 import static com.google.cloud.RetryHelper.runWithRetries;
+import static java.util.Objects.requireNonNull;
 import static java.util.concurrent.Executors.callable;
 
 import com.google.api.services.storage.model.StorageObject;
 import com.google.cloud.BaseWriteChannel;
+import com.google.cloud.ExceptionHandler;
 import com.google.cloud.RestorableState;
 import com.google.cloud.RetryHelper;
 import com.google.cloud.WriteChannel;
-import com.google.cloud.storage.spi.v1.StorageRpc;
 import java.math.BigInteger;
-import java.net.URL;
-import java.util.Map;
-import java.util.concurrent.Callable;
+import java.util.function.Supplier;
+import org.checkerframework.checker.nullness.qual.NonNull;
 
 /** Write channel implementation to upload Google Cloud Storage blobs. */
 class BlobWriteChannel extends BaseWriteChannel<StorageOptions, BlobInfo> {
 
-  BlobWriteChannel(StorageOptions options, BlobInfo blob, Map<StorageRpc.Option, ?> optionsMap) {
-    this(options, blob, open(options, blob, optionsMap));
-  }
-
-  BlobWriteChannel(StorageOptions options, URL signedURL) {
-    this(options, open(signedURL, options));
-  }
-
-  BlobWriteChannel(StorageOptions options, BlobInfo blobInfo, String uploadId) {
-    super(options, blobInfo, uploadId);
-  }
-
-  BlobWriteChannel(StorageOptions options, String uploadId) {
-    super(options, null, uploadId);
-  }
-
-  // Contains metadata of the updated object or null if upload is not completed.
-  private StorageObject storageObject;
-
+  private final ExceptionHandler exceptionHandlerForWrite;
   // Detect if flushBuffer() is being retried or not.
   // TODO: I don't think this is thread safe, and there's probably a better way to detect a retry
   // occuring.
   private boolean retrying = false;
   private boolean checkingForLastChunk = false;
 
+  // Contains metadata of the updated object or null if upload is not completed.
+  private StorageObject storageObject;
+
+  BlobWriteChannel(
+      StorageOptions storageOptions,
+      BlobInfo blobInfo,
+      String uploadId,
+      ExceptionHandler exceptionHandlerForWrite) {
+    super(storageOptions, blobInfo, uploadId);
+    this.exceptionHandlerForWrite = exceptionHandlerForWrite;
+  }
+
   boolean isRetrying() {
     return retrying;
   }
@@ -275,7 +269,7 @@ class BlobWriteChannel extends BaseWriteChannel<StorageOptions, BlobInfo> {
                 }
               }),
           getOptions().getRetrySettings(),
-          StorageImpl.EXCEPTION_HANDLER,
+          exceptionHandlerForWrite,
           getOptions().getClock());
     } catch (RetryHelper.RetryHelperException e) {
       throw StorageException.translateAndThrow(e);
@@ -283,83 +277,73 @@ class BlobWriteChannel extends BaseWriteChannel<StorageOptions, BlobInfo> {
   }
 
   protected StateImpl.Builder stateBuilder() {
-    return StateImpl.builder(getOptions(), getEntity(), getUploadId());
+    return StateImpl.builder(getOptions(), getEntity(), getUploadId())
+        .setResultRetryAlgorithm(exceptionHandlerForWrite);
   }
 
-  private static String open(
-      final StorageOptions options,
-      final BlobInfo blob,
-      final Map<StorageRpc.Option, ?> optionsMap) {
-    try {
-      return runWithRetries(
-          new Callable<String>() {
-            @Override
-            public String call() {
-              return options.getStorageRpcV1().open(blob.toPb(), optionsMap);
-            }
-          },
-          options.getRetrySettings(),
-          StorageImpl.EXCEPTION_HANDLER,
-          options.getClock());
-    } catch (RetryHelper.RetryHelperException e) {
-      throw StorageException.translateAndThrow(e);
-    }
+  static Builder newBuilder() {
+    return new Builder();
   }
 
-  private static String open(final URL signedURL, final StorageOptions options) {
-    try {
-      return runWithRetries(
-          new Callable<String>() {
-            @Override
-            public String call() {
-              if (!isValidSignedURL(signedURL.getQuery())) {
-                throw new StorageException(2, "invalid signedURL");
-              }
-              return options.getStorageRpcV1().open(signedURL.toString());
-            }
-          },
-          options.getRetrySettings(),
-          StorageImpl.EXCEPTION_HANDLER,
-          options.getClock());
-    } catch (RetryHelper.RetryHelperException e) {
-      throw StorageException.translateAndThrow(e);
+  static final class Builder {
+    private StorageOptions storageOptions;
+    private BlobInfo blobInfo;
+    private Supplier<@NonNull String> uploadIdSupplier;
+    private ExceptionHandler putExceptionHandler;
+
+    public Builder setStorageOptions(StorageOptions storageOptions) {
+      this.storageOptions = storageOptions;
+      return this;
     }
-  }
 
-  private static boolean isValidSignedURL(String signedURLQuery) {
-    boolean isValid = true;
-    if (signedURLQuery.startsWith("X-Goog-Algorithm=")) {
-      if (!signedURLQuery.contains("&X-Goog-Credential=")
-          || !signedURLQuery.contains("&X-Goog-Date=")
-          || !signedURLQuery.contains("&X-Goog-Expires=")
-          || !signedURLQuery.contains("&X-Goog-SignedHeaders=")
-          || !signedURLQuery.contains("&X-Goog-Signature=")) {
-        isValid = false;
-      }
-    } else if (signedURLQuery.startsWith("GoogleAccessId=")) {
-      if (!signedURLQuery.contains("&Expires=") || !signedURLQuery.contains("&Signature=")) {
-        isValid = false;
-      }
-    } else {
-      isValid = false;
+    public Builder setBlobInfo(BlobInfo blobInfo) {
+      this.blobInfo = blobInfo;
+      return this;
+    }
+
+    public Builder setUploadIdSupplier(Supplier<String> uploadIdSupplier) {
+      this.uploadIdSupplier = uploadIdSupplier;
+      return this;
+    }
+
+    public Builder setPutExceptionHandler(ExceptionHandler putExceptionHandler) {
+      this.putExceptionHandler = putExceptionHandler;
+      return this;
+    }
+
+    BlobWriteChannel build() {
+      String uploadId = requireNonNull(uploadIdSupplier, "uploadId must be non null").get();
+      return new BlobWriteChannel(
+          requireNonNull(storageOptions, "storageOptions must be non null"),
+          blobInfo,
+          requireNonNull(uploadId, "uploadId must be non null"),
+          requireNonNull(putExceptionHandler, "putExceptionHandler must be non null"));
     }
-    return isValid;
   }
 
   static class StateImpl extends BaseWriteChannel.BaseState<StorageOptions, BlobInfo> {
 
     private static final long serialVersionUID = -9028324143780151286L;
 
+    private final ExceptionHandler exceptionHandler;
+
     StateImpl(Builder builder) {
       super(builder);
+      this.exceptionHandler = builder.exceptionHandler;
     }
 
     static class Builder extends BaseWriteChannel.BaseState.Builder<StorageOptions, BlobInfo> {
+      private ExceptionHandler exceptionHandler;
 
       private Builder(StorageOptions options, BlobInfo blobInfo, String uploadId) {
         super(options, blobInfo, uploadId);
       }
 
+      public Builder setResultRetryAlgorithm(ExceptionHandler exceptionHandler) {
+        this.exceptionHandler = exceptionHandler;
+        return this;
+      }
+
       @Override
       public RestorableState<WriteChannel> build() {
         return new StateImpl(this);
@@ -372,9 +356,19 @@ class BlobWriteChannel extends BaseWriteChannel<StorageOptions, BlobInfo> {
 
     @Override
     public WriteChannel restore() {
-      BlobWriteChannel channel = new BlobWriteChannel(serviceOptions, entity, uploadId);
-      channel.restore(this);
-      return channel;
+      try {
+        BlobWriteChannel channel =
+            BlobWriteChannel.newBuilder()
+                .setStorageOptions(serviceOptions)
+                .setBlobInfo(entity)
+                .setUploadIdSupplier(() -> uploadId)
+                .setPutExceptionHandler(exceptionHandler)
+                .build();
+        channel.restore(this);
+        return channel;
+      } catch (Exception e) {
+        throw StorageException.coalesce(e);
+      }
     }
   }
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/CopyWriter.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/CopyWriter.java
index 47aa6b9f..f3dcdfde 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/CopyWriter.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/CopyWriter.java
@@ -100,6 +100,7 @@ public class CopyWriter implements Restorable<CopyWriter> {
       this.rewriteResponse =
           Retrying.run(
               serviceOptions,
+              serviceOptions.getRetryAlgorithmManager().getForObjectsCopy(),
               () -> storageRpc.continueRewrite(rewriteResponse),
               Function.identity());
     }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/LegacyRetryAlgorithmManager.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/LegacyRetryAlgorithmManager.java
new file mode 100644
index 00000000..fd88c333
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/LegacyRetryAlgorithmManager.java
@@ -0,0 +1,247 @@
+/*
+ * Copyright 2021 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.storage;
+
+import com.google.api.services.storage.model.Bucket;
+import com.google.api.services.storage.model.BucketAccessControl;
+import com.google.api.services.storage.model.HmacKeyMetadata;
+import com.google.api.services.storage.model.ObjectAccessControl;
+import com.google.api.services.storage.model.Policy;
+import com.google.api.services.storage.model.StorageObject;
+import com.google.cloud.BaseService;
+import com.google.cloud.ExceptionHandler;
+import com.google.cloud.storage.spi.v1.StorageRpc;
+import com.google.cloud.storage.spi.v1.StorageRpc.RewriteRequest;
+import java.util.List;
+import java.util.Map;
+
+class LegacyRetryAlgorithmManager implements RetryAlgorithmManager {
+
+  @Override
+  public ExceptionHandler getForBucketAclCreate(
+      BucketAccessControl pb, Map<StorageRpc.Option, ?> optionsMap) {
+    return BaseService.EXCEPTION_HANDLER;
+  }
+
+  @Override
+  public ExceptionHandler getForBucketAclDelete(String pb, Map<StorageRpc.Option, ?> optionsMap) {
+    return BaseService.EXCEPTION_HANDLER;
+  }
+
+  @Override
+  public ExceptionHandler getForBucketAclGet(String pb, Map<StorageRpc.Option, ?> optionsMap) {
+    return BaseService.EXCEPTION_HANDLER;
+  }
+
+  @Override
+  public ExceptionHandler getForBucketAclUpdate(
+      BucketAccessControl pb, Map<StorageRpc.Option, ?> optionsMap) {
+    return BaseService.EXCEPTION_HANDLER;
+  }
+
+  @Override
+  public ExceptionHandler getForBucketAclList(String pb, Map<StorageRpc.Option, ?> optionsMap) {
+    return BaseService.EXCEPTION_HANDLER;
+  }
+
+  @Override
+  public ExceptionHandler getForBucketsCreate(Bucket pb, Map<StorageRpc.Option, ?> optionsMap) {
+    return BaseService.EXCEPTION_HANDLER;
+  }
+
+  @Override
+  public ExceptionHandler getForBucketsDelete(Bucket pb, Map<StorageRpc.Option, ?> optionsMap) {
+    return BaseService.EXCEPTION_HANDLER;
+  }
+
+  @Override
+  public ExceptionHandler getForBucketsGet(Bucket pb, Map<StorageRpc.Option, ?> optionsMap) {
+    return BaseService.EXCEPTION_HANDLER;
+  }
+
+  @Override
+  public ExceptionHandler getForBucketsUpdate(Bucket pb, Map<StorageRpc.Option, ?> optionsMap) {
+    return BaseService.EXCEPTION_HANDLER;
+  }
+
+  @Override
+  public ExceptionHandler getForBucketsList(Map<StorageRpc.Option, ?> optionsMap) {
+    return BaseService.EXCEPTION_HANDLER;
+  }
+
+  @Override
+  public ExceptionHandler getForBucketsLockRetentionPolicy(
+      Bucket pb, Map<StorageRpc.Option, ?> optionsMap) {
+    return BaseService.EXCEPTION_HANDLER;
+  }
+
+  @Override
+  public ExceptionHandler getForBucketsGetIamPolicy(
+      String bucket, Map<StorageRpc.Option, ?> optionsMap) {
+    return BaseService.EXCEPTION_HANDLER;
+  }
+
+  @Override
+  public ExceptionHandler getForBucketsSetIamPolicy(
+      String bucket, Policy pb, Map<StorageRpc.Option, ?> optionsMap) {
+    return BaseService.EXCEPTION_HANDLER;
+  }
+
+  @Override
+  public ExceptionHandler getForBucketsTestIamPermissions(
+      String bucket, List<String> permissions, Map<StorageRpc.Option, ?> optionsMap) {
+    return BaseService.EXCEPTION_HANDLER;
+  }
+
+  @Override
+  public ExceptionHandler getForDefaultObjectAclCreate(ObjectAccessControl pb) {
+    return BaseService.EXCEPTION_HANDLER;
+  }
+
+  @Override
+  public ExceptionHandler getForDefaultObjectAclDelete(String pb) {
+    return BaseService.EXCEPTION_HANDLER;
+  }
+
+  @Override
+  public ExceptionHandler getForDefaultObjectAclGet(String pb) {
+    return BaseService.EXCEPTION_HANDLER;
+  }
+
+  @Override
+  public ExceptionHandler getForDefaultObjectAclUpdate(ObjectAccessControl pb) {
+    return BaseService.EXCEPTION_HANDLER;
+  }
+
+  @Override
+  public ExceptionHandler getForDefaultObjectAclList(String pb) {
+    return BaseService.EXCEPTION_HANDLER;
+  }
+
+  @Override
+  public ExceptionHandler getForHmacKeyCreate(String pb, Map<StorageRpc.Option, ?> optionsMap) {
+    return BaseService.EXCEPTION_HANDLER;
+  }
+
+  @Override
+  public ExceptionHandler getForHmacKeyDelete(
+      HmacKeyMetadata pb, Map<StorageRpc.Option, ?> optionsMap) {
+    return BaseService.EXCEPTION_HANDLER;
+  }
+
+  @Override
+  public ExceptionHandler getForHmacKeyGet(String accessId, Map<StorageRpc.Option, ?> optionsMap) {
+    return BaseService.EXCEPTION_HANDLER;
+  }
+
+  @Override
+  public ExceptionHandler getForHmacKeyUpdate(
+      HmacKeyMetadata pb, Map<StorageRpc.Option, ?> optionsMap) {
+    return BaseService.EXCEPTION_HANDLER;
+  }
+
+  @Override
+  public ExceptionHandler getForHmacKeyList(Map<StorageRpc.Option, ?> optionsMap) {
+    return BaseService.EXCEPTION_HANDLER;
+  }
+
+  @Override
+  public ExceptionHandler getForObjectAclCreate(ObjectAccessControl aclPb) {
+    return BaseService.EXCEPTION_HANDLER;
+  }
+
+  @Override
+  public ExceptionHandler getForObjectAclDelete(
+      String bucket, String name, Long generation, String pb) {
+    return BaseService.EXCEPTION_HANDLER;
+  }
+
+  @Override
+  public ExceptionHandler getForObjectAclList(String bucket, String name, Long generation) {
+    return BaseService.EXCEPTION_HANDLER;
+  }
+
+  @Override
+  public ExceptionHandler getForObjectAclGet(
+      String bucket, String name, Long generation, String pb) {
+    return BaseService.EXCEPTION_HANDLER;
+  }
+
+  @Override
+  public ExceptionHandler getForObjectAclUpdate(ObjectAccessControl aclPb) {
+    return BaseService.EXCEPTION_HANDLER;
+  }
+
+  @Override
+  public ExceptionHandler getForObjectsCreate(
+      StorageObject pb, Map<StorageRpc.Option, ?> optionsMap) {
+    return BaseService.EXCEPTION_HANDLER;
+  }
+
+  @Override
+  public ExceptionHandler getForObjectsDelete(
+      StorageObject pb, Map<StorageRpc.Option, ?> optionsMap) {
+    return BaseService.EXCEPTION_HANDLER;
+  }
+
+  @Override
+  public ExceptionHandler getForObjectsGet(StorageObject pb, Map<StorageRpc.Option, ?> optionsMap) {
+    return BaseService.EXCEPTION_HANDLER;
+  }
+
+  @Override
+  public ExceptionHandler getForObjectsUpdate(
+      StorageObject pb, Map<StorageRpc.Option, ?> optionsMap) {
+    return BaseService.EXCEPTION_HANDLER;
+  }
+
+  @Override
+  public ExceptionHandler getForObjectsList(String bucket, Map<StorageRpc.Option, ?> optionsMap) {
+    return BaseService.EXCEPTION_HANDLER;
+  }
+
+  @Override
+  public ExceptionHandler getForObjectsRewrite(RewriteRequest pb) {
+    return BaseService.EXCEPTION_HANDLER;
+  }
+
+  @Override
+  public ExceptionHandler getForObjectsCopy() {
+    return BaseService.EXCEPTION_HANDLER;
+  }
+
+  @Override
+  public ExceptionHandler getForObjectsCompose(
+      List<StorageObject> sources, StorageObject target, Map<StorageRpc.Option, ?> targetOptions) {
+    return BaseService.EXCEPTION_HANDLER;
+  }
+
+  @Override
+  public ExceptionHandler getForResumableUploadSessionCreate(Map<StorageRpc.Option, ?> optionsMap) {
+    return BaseService.EXCEPTION_HANDLER;
+  }
+
+  @Override
+  public ExceptionHandler getForResumableUploadSessionWrite(Map<StorageRpc.Option, ?> optionsMap) {
+    return BaseService.EXCEPTION_HANDLER;
+  }
+
+  @Override
+  public ExceptionHandler getForServiceAccountGet(String pb) {
+    return BaseService.EXCEPTION_HANDLER;
+  }
+}
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/NewRetryAlgorithmManager.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/NewRetryAlgorithmManager.java
new file mode 100644
index 00000000..0c3e1f8e
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/NewRetryAlgorithmManager.java
@@ -0,0 +1,20 @@
+/*
+ * Copyright 2021 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.storage;
+
+final class NewRetryAlgorithmManager extends LegacyRetryAlgorithmManager
+    implements RetryAlgorithmManager {}
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/ResumableMedia.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/ResumableMedia.java
new file mode 100644
index 00000000..d0e212ef
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/ResumableMedia.java
@@ -0,0 +1,74 @@
+/*
+ * Copyright 2021 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.storage;
+
+import com.google.api.core.ApiFutures;
+import com.google.cloud.ExceptionHandler;
+import com.google.cloud.storage.spi.v1.StorageRpc;
+import java.net.URL;
+import java.util.Map;
+import java.util.function.Function;
+import java.util.function.Supplier;
+
+final class ResumableMedia {
+
+  static Supplier<String> startUploadForBlobInfo(
+      final StorageOptions storageOptions,
+      final BlobInfo blob,
+      final Map<StorageRpc.Option, ?> optionsMap,
+      ExceptionHandler exceptionHandler) {
+    return () ->
+        Retrying.run(
+            storageOptions,
+            exceptionHandler,
+            () -> storageOptions.getStorageRpcV1().open(blob.toPb(), optionsMap),
+            Function.identity());
+  }
+
+  static Supplier<String> startUploadForSignedUrl(
+      final StorageOptions storageOptions, final URL signedURL, ExceptionHandler exceptionHandler) {
+    if (!isValidSignedURL(signedURL.getQuery())) {
+      ApiFutures.immediateFailedFuture(new StorageException(2, "invalid signedURL"));
+    }
+    return () ->
+        Retrying.run(
+            storageOptions,
+            exceptionHandler,
+            () -> storageOptions.getStorageRpcV1().open(signedURL.toString()),
+            Function.identity());
+  }
+
+  private static boolean isValidSignedURL(String signedURLQuery) {
+    boolean isValid = true;
+    if (signedURLQuery.startsWith("X-Goog-Algorithm=")) {
+      if (!signedURLQuery.contains("&X-Goog-Credential=")
+          || !signedURLQuery.contains("&X-Goog-Date=")
+          || !signedURLQuery.contains("&X-Goog-Expires=")
+          || !signedURLQuery.contains("&X-Goog-SignedHeaders=")
+          || !signedURLQuery.contains("&X-Goog-Signature=")) {
+        isValid = false;
+      }
+    } else if (signedURLQuery.startsWith("GoogleAccessId=")) {
+      if (!signedURLQuery.contains("&Expires=") || !signedURLQuery.contains("&Signature=")) {
+        isValid = false;
+      }
+    } else {
+      isValid = false;
+    }
+    return isValid;
+  }
+}
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/RetryAlgorithmManager.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/RetryAlgorithmManager.java
new file mode 100644
index 00000000..884c4d4e
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/RetryAlgorithmManager.java
@@ -0,0 +1,118 @@
+/*
+ * Copyright 2021 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.storage;
+
+import com.google.api.services.storage.model.Bucket;
+import com.google.api.services.storage.model.BucketAccessControl;
+import com.google.api.services.storage.model.HmacKeyMetadata;
+import com.google.api.services.storage.model.ObjectAccessControl;
+import com.google.api.services.storage.model.Policy;
+import com.google.api.services.storage.model.StorageObject;
+import com.google.cloud.ExceptionHandler;
+import com.google.cloud.storage.spi.v1.StorageRpc;
+import com.google.cloud.storage.spi.v1.StorageRpc.RewriteRequest;
+import java.io.Serializable;
+import java.util.List;
+import java.util.Map;
+
+interface RetryAlgorithmManager extends Serializable {
+  ExceptionHandler getForBucketAclCreate(
+      BucketAccessControl pb, Map<StorageRpc.Option, ?> optionsMap);
+
+  ExceptionHandler getForBucketAclDelete(String pb, Map<StorageRpc.Option, ?> optionsMap);
+
+  ExceptionHandler getForBucketAclGet(String pb, Map<StorageRpc.Option, ?> optionsMap);
+
+  ExceptionHandler getForBucketAclUpdate(
+      BucketAccessControl pb, Map<StorageRpc.Option, ?> optionsMap);
+
+  ExceptionHandler getForBucketAclList(String pb, Map<StorageRpc.Option, ?> optionsMap);
+
+  ExceptionHandler getForBucketsCreate(Bucket pb, Map<StorageRpc.Option, ?> optionsMap);
+
+  ExceptionHandler getForBucketsDelete(Bucket pb, Map<StorageRpc.Option, ?> optionsMap);
+
+  ExceptionHandler getForBucketsGet(Bucket pb, Map<StorageRpc.Option, ?> optionsMap);
+
+  ExceptionHandler getForBucketsUpdate(Bucket pb, Map<StorageRpc.Option, ?> optionsMap);
+
+  ExceptionHandler getForBucketsList(Map<StorageRpc.Option, ?> optionsMap);
+
+  ExceptionHandler getForBucketsLockRetentionPolicy(
+      Bucket pb, Map<StorageRpc.Option, ?> optionsMap);
+
+  ExceptionHandler getForBucketsGetIamPolicy(String bucket, Map<StorageRpc.Option, ?> optionsMap);
+
+  ExceptionHandler getForBucketsSetIamPolicy(
+      String bucket, Policy pb, Map<StorageRpc.Option, ?> optionsMap);
+
+  ExceptionHandler getForBucketsTestIamPermissions(
+      String bucket, List<String> permissions, Map<StorageRpc.Option, ?> optionsMap);
+
+  ExceptionHandler getForDefaultObjectAclCreate(ObjectAccessControl pb);
+
+  ExceptionHandler getForDefaultObjectAclDelete(String pb);
+
+  ExceptionHandler getForDefaultObjectAclGet(String pb);
+
+  ExceptionHandler getForDefaultObjectAclUpdate(ObjectAccessControl pb);
+
+  ExceptionHandler getForDefaultObjectAclList(String pb);
+
+  ExceptionHandler getForHmacKeyCreate(String pb, Map<StorageRpc.Option, ?> optionsMap);
+
+  ExceptionHandler getForHmacKeyDelete(HmacKeyMetadata pb, Map<StorageRpc.Option, ?> optionsMap);
+
+  ExceptionHandler getForHmacKeyGet(String accessId, Map<StorageRpc.Option, ?> optionsMap);
+
+  ExceptionHandler getForHmacKeyUpdate(HmacKeyMetadata pb, Map<StorageRpc.Option, ?> optionsMap);
+
+  ExceptionHandler getForHmacKeyList(Map<StorageRpc.Option, ?> optionsMap);
+
+  ExceptionHandler getForObjectAclCreate(ObjectAccessControl aclPb);
+
+  ExceptionHandler getForObjectAclDelete(String bucket, String name, Long generation, String pb);
+
+  ExceptionHandler getForObjectAclList(String bucket, String name, Long generation);
+
+  ExceptionHandler getForObjectAclGet(String bucket, String name, Long generation, String pb);
+
+  ExceptionHandler getForObjectAclUpdate(ObjectAccessControl aclPb);
+
+  ExceptionHandler getForObjectsCreate(StorageObject pb, Map<StorageRpc.Option, ?> optionsMap);
+
+  ExceptionHandler getForObjectsDelete(StorageObject pb, Map<StorageRpc.Option, ?> optionsMap);
+
+  ExceptionHandler getForObjectsGet(StorageObject pb, Map<StorageRpc.Option, ?> optionsMap);
+
+  ExceptionHandler getForObjectsUpdate(StorageObject pb, Map<StorageRpc.Option, ?> optionsMap);
+
+  ExceptionHandler getForObjectsList(String bucket, Map<StorageRpc.Option, ?> optionsMap);
+
+  ExceptionHandler getForObjectsRewrite(RewriteRequest pb);
+
+  ExceptionHandler getForObjectsCopy();
+
+  ExceptionHandler getForObjectsCompose(
+      List<StorageObject> sources, StorageObject target, Map<StorageRpc.Option, ?> targetOptions);
+
+  ExceptionHandler getForResumableUploadSessionCreate(Map<StorageRpc.Option, ?> optionsMap);
+  /** Resumable upload has differing 429 handling */
+  ExceptionHandler getForResumableUploadSessionWrite(Map<StorageRpc.Option, ?> optionsMap);
+
+  ExceptionHandler getForServiceAccountGet(String pb);
+}
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/Retrying.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/Retrying.java
index 3cae94ea..603d43ae 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/Retrying.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/Retrying.java
@@ -21,7 +21,7 @@ import static com.google.cloud.RetryHelper.runWithRetries;
 import com.google.api.core.ApiClock;
 import com.google.api.gax.retrying.ResultRetryAlgorithm;
 import com.google.api.gax.retrying.RetrySettings;
-import com.google.cloud.BaseService;
+import com.google.cloud.ExceptionHandler;
 import com.google.cloud.RetryHelper.RetryHelperException;
 import java.util.concurrent.Callable;
 import java.util.function.Function;
@@ -33,25 +33,27 @@ final class Retrying {
    * RetrySettings, ResultRetryAlgorithm, ApiClock)} that gives us centralized error translation and
    * reduces some duplication in how we resolved the {@link RetrySettings} and {@link ApiClock}.
    *
+   * @param <T> The result type of {@code c}
+   * @param <U> The result type of any mapping that takes place via {@code f}
    * @param options The {@link StorageOptions} which {@link RetrySettings} and {@link ApiClock} will
    *     be resolved from.
+   * @param exceptionHandler The {@link ExceptionHandler} to use when determining if a retry is
+   *     possible
    * @param c The {@link Callable} which will be passed to runWithRetries producing some {@code T},
    *     can optionally return null
    * @param f A post process mapping {@link Function} which can be used to transform the result from
    *     {@code c} if it is successful and non-null
-   * @param <T> The result type of {@code c}
-   * @param <U> The result type of any mapping that takes place via {@code f}
    * @return A {@code U} (possibly null) after applying {@code f} to the result of {@code c}
    * @throws StorageException if {@code c} fails due to any retry exhaustion
    */
-  static <T, U> U run(StorageOptions options, Callable<T> c, Function<T, U> f) {
+  static <T, U> U run(
+      StorageOptions options, ExceptionHandler exceptionHandler, Callable<T> c, Function<T, U> f) {
     try {
-      T answer =
-          runWithRetries(
-              c, options.getRetrySettings(), BaseService.EXCEPTION_HANDLER, options.getClock());
-      return answer == null ? null : f.apply(answer);
+      T result =
+          runWithRetries(c, options.getRetrySettings(), exceptionHandler, options.getClock());
+      return result == null ? null : f.apply(result);
     } catch (RetryHelperException e) {
-      throw StorageException.translateAndThrow(e);
+      throw StorageException.coalesce(e);
     }
   }
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageException.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageException.java
index f4b8d0c4..d0b3e9c7 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageException.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageException.java
@@ -78,7 +78,26 @@ public final class StorageException extends BaseHttpServiceException {
    */
   public static StorageException translateAndThrow(RetryHelperException ex) {
     BaseServiceException.translate(ex);
-    throw new StorageException(UNKNOWN_CODE, ex.getMessage(), ex.getCause());
+    throw getStorageException(ex);
+  }
+
+  private static StorageException getStorageException(Throwable t) {
+    return new StorageException(UNKNOWN_CODE, t.getMessage(), t.getCause());
+  }
+
+  /**
+   * Attempt to find an Exception which is a {@link BaseServiceException} If neither {@code t} or
+   * {@code t.getCause()} are a {@code BaseServiceException} a {@link StorageException} will be
+   * created with an unknown status code.
+   */
+  static BaseServiceException coalesce(Throwable t) {
+    if (t instanceof BaseServiceException) {
+      return (BaseServiceException) t;
+    }
+    if (t.getCause() instanceof BaseServiceException) {
+      return (BaseServiceException) t.getCause();
+    }
+    return getStorageException(t);
   }
 
   /**
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageImpl.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageImpl.java
index 2d7f9675..483387a0 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageImpl.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageImpl.java
@@ -39,6 +39,7 @@ import com.google.api.services.storage.model.StorageObject;
 import com.google.auth.ServiceAccountSigner;
 import com.google.cloud.BaseService;
 import com.google.cloud.BatchResult;
+import com.google.cloud.ExceptionHandler;
 import com.google.cloud.PageImpl;
 import com.google.cloud.PageImpl.NextPageFetcher;
 import com.google.cloud.Policy;
@@ -52,6 +53,7 @@ import com.google.cloud.storage.PostPolicyV4.PostConditionsV4;
 import com.google.cloud.storage.PostPolicyV4.PostFieldsV4;
 import com.google.cloud.storage.PostPolicyV4.PostPolicyV4Document;
 import com.google.cloud.storage.spi.v1.StorageRpc;
+import com.google.cloud.storage.spi.v1.StorageRpc.RewriteRequest;
 import com.google.common.base.CharMatcher;
 import com.google.common.base.Preconditions;
 import com.google.common.base.Strings;
@@ -104,18 +106,25 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
   private static final int DEFAULT_BUFFER_SIZE = 15 * 1024 * 1024;
   private static final int MIN_BUFFER_SIZE = 256 * 1024;
 
+  private final RetryAlgorithmManager retryAlgorithmManager;
   private final StorageRpc storageRpc;
 
   StorageImpl(StorageOptions options) {
     super(options);
-    storageRpc = options.getStorageRpcV1();
+    this.retryAlgorithmManager = options.getRetryAlgorithmManager();
+    this.storageRpc = options.getStorageRpcV1();
   }
 
   @Override
   public Bucket create(BucketInfo bucketInfo, BucketTargetOption... options) {
     final com.google.api.services.storage.model.Bucket bucketPb = bucketInfo.toPb();
     final Map<StorageRpc.Option, ?> optionsMap = optionMap(bucketInfo, options);
-    return run(() -> storageRpc.create(bucketPb, optionsMap), (b) -> Bucket.fromPb(this, b));
+    ExceptionHandler exceptionHandler =
+        retryAlgorithmManager.getForBucketsCreate(bucketPb, optionsMap);
+    return run(
+        exceptionHandler,
+        () -> storageRpc.create(bucketPb, optionsMap),
+        (b) -> Bucket.fromPb(this, b));
   }
 
   @Override
@@ -183,7 +192,10 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
     Preconditions.checkNotNull(content);
     final StorageObject blobPb = info.toPb();
     final Map<StorageRpc.Option, ?> optionsMap = optionMap(info, options);
+    ExceptionHandler exceptionHandler =
+        retryAlgorithmManager.getForObjectsCreate(blobPb, optionsMap);
     return run(
+        exceptionHandler,
         () ->
             storageRpc.create(
                 blobPb, new ByteArrayInputStream(content, offset, length), optionsMap),
@@ -248,7 +260,12 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
   public Bucket get(String bucket, BucketGetOption... options) {
     final com.google.api.services.storage.model.Bucket bucketPb = BucketInfo.of(bucket).toPb();
     final Map<StorageRpc.Option, ?> optionsMap = optionMap(options);
-    return run(() -> storageRpc.get(bucketPb, optionsMap), (b) -> Bucket.fromPb(this, b));
+    ExceptionHandler exceptionHandler =
+        retryAlgorithmManager.getForBucketsGet(bucketPb, optionsMap);
+    return run(
+        exceptionHandler,
+        () -> storageRpc.get(bucketPb, optionsMap),
+        (b) -> Bucket.fromPb(this, b));
   }
 
   @Override
@@ -260,7 +277,12 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
   public Blob get(BlobId blob, BlobGetOption... options) {
     final StorageObject storedObject = blob.toPb();
     final Map<StorageRpc.Option, ?> optionsMap = optionMap(blob, options);
-    return run(() -> storageRpc.get(storedObject, optionsMap), (x) -> Blob.fromPb(this, x));
+    ExceptionHandler exceptionHandler =
+        retryAlgorithmManager.getForObjectsGet(storedObject, optionsMap);
+    return run(
+        exceptionHandler,
+        () -> storageRpc.get(storedObject, optionsMap),
+        (x) -> Blob.fromPb(this, x));
   }
 
   @Override
@@ -315,16 +337,21 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
 
     private static final long serialVersionUID = 308012320541700881L;
     private final StorageOptions serviceOptions;
+    private final RetryAlgorithmManager retryAlgorithmManager;
     private final Map<StorageRpc.Option, ?> options;
 
-    HmacKeyMetadataPageFetcher(StorageOptions serviceOptions, Map<StorageRpc.Option, ?> options) {
+    HmacKeyMetadataPageFetcher(
+        StorageOptions serviceOptions,
+        RetryAlgorithmManager retryAlgorithmManager,
+        Map<StorageRpc.Option, ?> options) {
       this.serviceOptions = serviceOptions;
+      this.retryAlgorithmManager = retryAlgorithmManager;
       this.options = options;
     }
 
     @Override
     public Page<HmacKeyMetadata> getNextPage() {
-      return listHmacKeys(serviceOptions, options);
+      return listHmacKeys(serviceOptions, retryAlgorithmManager, options);
     }
   }
 
@@ -340,8 +367,11 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
 
   private static Page<Bucket> listBuckets(
       final StorageOptions serviceOptions, final Map<StorageRpc.Option, ?> optionsMap) {
+    ExceptionHandler exceptionHandler =
+        serviceOptions.getRetryAlgorithmManager().getForBucketsList(optionsMap);
     return Retrying.run(
         serviceOptions,
+        exceptionHandler,
         () -> serviceOptions.getStorageRpcV1().list(optionsMap),
         (result) -> {
           String cursor = result.x();
@@ -359,8 +389,11 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
       final String bucket,
       final StorageOptions serviceOptions,
       final Map<StorageRpc.Option, ?> optionsMap) {
+    ExceptionHandler exceptionHandler =
+        serviceOptions.getRetryAlgorithmManager().getForObjectsList(bucket, optionsMap);
     return Retrying.run(
         serviceOptions,
+        exceptionHandler,
         () -> serviceOptions.getStorageRpcV1().list(bucket, optionsMap),
         (result) -> {
           String cursor = result.x();
@@ -379,14 +412,24 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
   public Bucket update(BucketInfo bucketInfo, BucketTargetOption... options) {
     final com.google.api.services.storage.model.Bucket bucketPb = bucketInfo.toPb();
     final Map<StorageRpc.Option, ?> optionsMap = optionMap(bucketInfo, options);
-    return run(() -> storageRpc.patch(bucketPb, optionsMap), (x) -> Bucket.fromPb(this, x));
+    ExceptionHandler exceptionHandler =
+        retryAlgorithmManager.getForBucketsUpdate(bucketPb, optionsMap);
+    return run(
+        exceptionHandler,
+        () -> storageRpc.patch(bucketPb, optionsMap),
+        (x) -> Bucket.fromPb(this, x));
   }
 
   @Override
   public Blob update(BlobInfo blobInfo, BlobTargetOption... options) {
     final StorageObject storageObject = blobInfo.toPb();
     final Map<StorageRpc.Option, ?> optionsMap = optionMap(blobInfo, options);
-    return run(() -> storageRpc.patch(storageObject, optionsMap), (x) -> Blob.fromPb(this, x));
+    ExceptionHandler exceptionHandler =
+        retryAlgorithmManager.getForObjectsUpdate(storageObject, optionsMap);
+    return run(
+        exceptionHandler,
+        () -> storageRpc.patch(storageObject, optionsMap),
+        (x) -> Blob.fromPb(this, x));
   }
 
   @Override
@@ -398,7 +441,10 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
   public boolean delete(String bucket, BucketSourceOption... options) {
     final com.google.api.services.storage.model.Bucket bucketPb = BucketInfo.of(bucket).toPb();
     final Map<StorageRpc.Option, ?> optionsMap = optionMap(options);
-    return run(() -> storageRpc.delete(bucketPb, optionsMap), Function.identity());
+    ExceptionHandler exceptionHandler =
+        retryAlgorithmManager.getForBucketsDelete(bucketPb, optionsMap);
+    return run(
+        exceptionHandler, () -> storageRpc.delete(bucketPb, optionsMap), Function.identity());
   }
 
   @Override
@@ -410,7 +456,10 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
   public boolean delete(BlobId blob, BlobSourceOption... options) {
     final StorageObject storageObject = blob.toPb();
     final Map<StorageRpc.Option, ?> optionsMap = optionMap(blob, options);
-    return run(() -> storageRpc.delete(storageObject, optionsMap), Function.identity());
+    ExceptionHandler exceptionHandler =
+        retryAlgorithmManager.getForObjectsDelete(storageObject, optionsMap);
+    return run(
+        exceptionHandler, () -> storageRpc.delete(storageObject, optionsMap), Function.identity());
   }
 
   @Override
@@ -438,8 +487,12 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
             composeRequest.getTarget().getGeneration(),
             composeRequest.getTarget().getMetageneration(),
             composeRequest.getTargetOptions());
+    ExceptionHandler exceptionHandler =
+        retryAlgorithmManager.getForObjectsCompose(sources, target, targetOptions);
     return run(
-        () -> storageRpc.compose(sources, target, targetOptions), (x) -> Blob.fromPb(this, x));
+        exceptionHandler,
+        () -> storageRpc.compose(sources, target, targetOptions),
+        (x) -> Blob.fromPb(this, x));
   }
 
   @Override
@@ -454,16 +507,18 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
             copyRequest.getTarget().getGeneration(),
             copyRequest.getTarget().getMetageneration(),
             copyRequest.getTargetOptions());
+    RewriteRequest rewriteRequest =
+        new RewriteRequest(
+            source,
+            sourceOptions,
+            copyRequest.overrideInfo(),
+            targetObject,
+            targetOptions,
+            copyRequest.getMegabytesCopiedPerChunk());
+    ExceptionHandler exceptionHandler = retryAlgorithmManager.getForObjectsRewrite(rewriteRequest);
     return run(
-        () ->
-            storageRpc.openRewrite(
-                new StorageRpc.RewriteRequest(
-                    source,
-                    sourceOptions,
-                    copyRequest.overrideInfo(),
-                    targetObject,
-                    targetOptions,
-                    copyRequest.getMegabytesCopiedPerChunk())),
+        exceptionHandler,
+        () -> storageRpc.openRewrite(rewriteRequest),
         (r) -> new CopyWriter(getOptions(), r));
   }
 
@@ -476,7 +531,10 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
   public byte[] readAllBytes(BlobId blob, BlobSourceOption... options) {
     final StorageObject storageObject = blob.toPb();
     final Map<StorageRpc.Option, ?> optionsMap = optionMap(blob, options);
-    return run(() -> storageRpc.load(storageObject, optionsMap), Function.identity());
+    ExceptionHandler exceptionHandler =
+        retryAlgorithmManager.getForObjectsGet(storageObject, optionsMap);
+    return run(
+        exceptionHandler, () -> storageRpc.load(storageObject, optionsMap), Function.identity());
   }
 
   @Override
@@ -504,12 +562,33 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
 
   @Override
   public BlobWriteChannel writer(URL signedURL) {
-    return new BlobWriteChannel(getOptions(), signedURL);
+    ExceptionHandler forResumableUploadSessionCreate =
+        retryAlgorithmManager.getForResumableUploadSessionCreate(
+            Collections
+                .emptyMap()); // TODO: is it possible to know if a signed url is configured to have
+    // a constraint which makes it idempotent?
+    return BlobWriteChannel.newBuilder()
+        .setStorageOptions(getOptions())
+        .setUploadIdSupplier(
+            ResumableMedia.startUploadForSignedUrl(
+                getOptions(), signedURL, forResumableUploadSessionCreate))
+        .setPutExceptionHandler(
+            retryAlgorithmManager.getForResumableUploadSessionWrite(optionMap()))
+        .build();
   }
 
   private BlobWriteChannel writer(BlobInfo blobInfo, BlobTargetOption... options) {
     final Map<StorageRpc.Option, ?> optionsMap = optionMap(blobInfo, options);
-    return new BlobWriteChannel(getOptions(), blobInfo, optionsMap);
+    return BlobWriteChannel.newBuilder()
+        .setStorageOptions(getOptions())
+        .setUploadIdSupplier(
+            ResumableMedia.startUploadForBlobInfo(
+                getOptions(),
+                blobInfo,
+                optionsMap,
+                retryAlgorithmManager.getForResumableUploadSessionCreate(optionsMap)))
+        .setPutExceptionHandler(retryAlgorithmManager.getForResumableUploadSessionWrite(optionsMap))
+        .build();
   }
 
   @Override
@@ -987,8 +1066,10 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
 
   @Override
   public Acl getAcl(final String bucket, final Entity entity, BucketSourceOption... options) {
+    String pb = entity.toPb();
     final Map<StorageRpc.Option, ?> optionsMap = optionMap(options);
-    return run(() -> storageRpc.getAcl(bucket, entity.toPb(), optionsMap), Acl::fromPb);
+    ExceptionHandler exceptionHandler = retryAlgorithmManager.getForBucketAclGet(pb, optionsMap);
+    return run(exceptionHandler, () -> storageRpc.getAcl(bucket, pb, optionsMap), Acl::fromPb);
   }
 
   @Override
@@ -999,8 +1080,11 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
   @Override
   public boolean deleteAcl(
       final String bucket, final Entity entity, BucketSourceOption... options) {
+    final String pb = entity.toPb();
     final Map<StorageRpc.Option, ?> optionsMap = optionMap(options);
-    return run(() -> storageRpc.deleteAcl(bucket, entity.toPb(), optionsMap), Function.identity());
+    ExceptionHandler exceptionHandler = retryAlgorithmManager.getForBucketAclDelete(pb, optionsMap);
+    return run(
+        exceptionHandler, () -> storageRpc.deleteAcl(bucket, pb, optionsMap), Function.identity());
   }
 
   @Override
@@ -1012,7 +1096,9 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
   public Acl createAcl(String bucket, Acl acl, BucketSourceOption... options) {
     final BucketAccessControl aclPb = acl.toBucketPb().setBucket(bucket);
     final Map<StorageRpc.Option, ?> optionsMap = optionMap(options);
-    return run(() -> storageRpc.createAcl(aclPb, optionsMap), Acl::fromPb);
+    ExceptionHandler exceptionHandler =
+        retryAlgorithmManager.getForBucketAclCreate(aclPb, optionsMap);
+    return run(exceptionHandler, () -> storageRpc.createAcl(aclPb, optionsMap), Acl::fromPb);
   }
 
   @Override
@@ -1024,7 +1110,9 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
   public Acl updateAcl(String bucket, Acl acl, BucketSourceOption... options) {
     final BucketAccessControl aclPb = acl.toBucketPb().setBucket(bucket);
     final Map<StorageRpc.Option, ?> optionsMap = optionMap(options);
-    return run(() -> storageRpc.patchAcl(aclPb, optionsMap), Acl::fromPb);
+    ExceptionHandler exceptionHandler =
+        retryAlgorithmManager.getForBucketAclUpdate(aclPb, optionsMap);
+    return run(exceptionHandler, () -> storageRpc.patchAcl(aclPb, optionsMap), Acl::fromPb);
   }
 
   @Override
@@ -1035,7 +1123,10 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
   @Override
   public List<Acl> listAcls(final String bucket, BucketSourceOption... options) {
     final Map<StorageRpc.Option, ?> optionsMap = optionMap(options);
+    ExceptionHandler exceptionHandler =
+        retryAlgorithmManager.getForBucketAclList(bucket, optionsMap);
     return run(
+        exceptionHandler,
         () -> storageRpc.listAcls(bucket, optionsMap),
         (answer) ->
             answer.stream()
@@ -1050,29 +1141,38 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
 
   @Override
   public Acl getDefaultAcl(final String bucket, final Entity entity) {
-    return run(() -> storageRpc.getDefaultAcl(bucket, entity.toPb()), Acl::fromPb);
+    String pb = entity.toPb();
+    ExceptionHandler exceptionHandler = retryAlgorithmManager.getForDefaultObjectAclGet(pb);
+    return run(exceptionHandler, () -> storageRpc.getDefaultAcl(bucket, pb), Acl::fromPb);
   }
 
   @Override
   public boolean deleteDefaultAcl(final String bucket, final Entity entity) {
-    return run(() -> storageRpc.deleteDefaultAcl(bucket, entity.toPb()), Function.identity());
+    String pb = entity.toPb();
+    ExceptionHandler exceptionHandler = retryAlgorithmManager.getForDefaultObjectAclDelete(pb);
+    return run(
+        exceptionHandler, () -> storageRpc.deleteDefaultAcl(bucket, pb), Function.identity());
   }
 
   @Override
   public Acl createDefaultAcl(String bucket, Acl acl) {
     final ObjectAccessControl aclPb = acl.toObjectPb().setBucket(bucket);
-    return run(() -> storageRpc.createDefaultAcl(aclPb), Acl::fromPb);
+    ExceptionHandler exceptionHandler = retryAlgorithmManager.getForDefaultObjectAclCreate(aclPb);
+    return run(exceptionHandler, () -> storageRpc.createDefaultAcl(aclPb), Acl::fromPb);
   }
 
   @Override
   public Acl updateDefaultAcl(String bucket, Acl acl) {
     final ObjectAccessControl aclPb = acl.toObjectPb().setBucket(bucket);
-    return run(() -> storageRpc.patchDefaultAcl(aclPb), Acl::fromPb);
+    ExceptionHandler exceptionHandler = retryAlgorithmManager.getForDefaultObjectAclUpdate(aclPb);
+    return run(exceptionHandler, () -> storageRpc.patchDefaultAcl(aclPb), Acl::fromPb);
   }
 
   @Override
   public List<Acl> listDefaultAcls(final String bucket) {
+    ExceptionHandler exceptionHandler = retryAlgorithmManager.getForDefaultObjectAclList(bucket);
     return run(
+        exceptionHandler,
         () -> storageRpc.listDefaultAcls(bucket),
         (answer) ->
             answer.stream()
@@ -1082,19 +1182,27 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
 
   @Override
   public Acl getAcl(final BlobId blob, final Entity entity) {
+    String bucket = blob.getBucket();
+    String name = blob.getName();
+    Long generation = blob.getGeneration();
+    String pb = entity.toPb();
+    ExceptionHandler exceptionHandler =
+        retryAlgorithmManager.getForObjectAclGet(bucket, name, generation, pb);
     return run(
-        () ->
-            storageRpc.getAcl(
-                blob.getBucket(), blob.getName(), blob.getGeneration(), entity.toPb()),
-        Acl::fromPb);
+        exceptionHandler, () -> storageRpc.getAcl(bucket, name, generation, pb), Acl::fromPb);
   }
 
   @Override
   public boolean deleteAcl(final BlobId blob, final Entity entity) {
+    String bucket = blob.getBucket();
+    String name = blob.getName();
+    Long generation = blob.getGeneration();
+    String pb = entity.toPb();
+    ExceptionHandler exceptionHandler =
+        retryAlgorithmManager.getForObjectAclDelete(bucket, name, generation, pb);
     return run(
-        () ->
-            storageRpc.deleteAcl(
-                blob.getBucket(), blob.getName(), blob.getGeneration(), entity.toPb()),
+        exceptionHandler,
+        () -> storageRpc.deleteAcl(bucket, name, generation, pb),
         Function.identity());
   }
 
@@ -1105,7 +1213,8 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
             .setBucket(blob.getBucket())
             .setObject(blob.getName())
             .setGeneration(blob.getGeneration());
-    return run(() -> storageRpc.createAcl(aclPb), Acl::fromPb);
+    ExceptionHandler exceptionHandler = retryAlgorithmManager.getForObjectAclCreate(aclPb);
+    return run(exceptionHandler, () -> storageRpc.createAcl(aclPb), Acl::fromPb);
   }
 
   @Override
@@ -1115,13 +1224,20 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
             .setBucket(blob.getBucket())
             .setObject(blob.getName())
             .setGeneration(blob.getGeneration());
-    return run(() -> storageRpc.patchAcl(aclPb), Acl::fromPb);
+    ExceptionHandler exceptionHandler = retryAlgorithmManager.getForObjectAclUpdate(aclPb);
+    return run(exceptionHandler, () -> storageRpc.patchAcl(aclPb), Acl::fromPb);
   }
 
   @Override
   public List<Acl> listAcls(final BlobId blob) {
+    String bucket = blob.getBucket();
+    String name = blob.getName();
+    Long generation = blob.getGeneration();
+    ExceptionHandler exceptionHandler =
+        retryAlgorithmManager.getForObjectAclList(bucket, name, generation);
     return run(
-        () -> storageRpc.listAcls(blob.getBucket(), blob.getName(), blob.getGeneration()),
+        exceptionHandler,
+        () -> storageRpc.listAcls(bucket, name, generation),
         (answer) ->
             answer.stream()
                 .map(Acl.FROM_OBJECT_PB_FUNCTION)
@@ -1130,26 +1246,35 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
 
   public HmacKey createHmacKey(
       final ServiceAccount serviceAccount, final CreateHmacKeyOption... options) {
-    return run(
-        () -> storageRpc.createHmacKey(serviceAccount.getEmail(), optionMap(options)),
-        HmacKey::fromPb);
+    String pb = serviceAccount.getEmail();
+    Map<StorageRpc.Option, ?> optionsMap = optionMap(options);
+    ExceptionHandler exceptionHandler = retryAlgorithmManager.getForHmacKeyCreate(pb, optionsMap);
+    return run(exceptionHandler, () -> storageRpc.createHmacKey(pb, optionsMap), HmacKey::fromPb);
   }
 
   @Override
   public Page<HmacKeyMetadata> listHmacKeys(ListHmacKeysOption... options) {
-    return listHmacKeys(getOptions(), optionMap(options));
+    return listHmacKeys(getOptions(), retryAlgorithmManager, optionMap(options));
   }
 
   @Override
   public HmacKeyMetadata getHmacKey(final String accessId, final GetHmacKeyOption... options) {
-    return run(() -> storageRpc.getHmacKey(accessId, optionMap(options)), HmacKeyMetadata::fromPb);
+    Map<StorageRpc.Option, ?> optionsMap = optionMap(options);
+    ExceptionHandler exceptionHandler =
+        retryAlgorithmManager.getForHmacKeyGet(accessId, optionsMap);
+    return run(
+        exceptionHandler,
+        () -> storageRpc.getHmacKey(accessId, optionMap(options)),
+        HmacKeyMetadata::fromPb);
   }
 
   private HmacKeyMetadata updateHmacKey(
       final HmacKeyMetadata hmacKeyMetadata, final UpdateHmacKeyOption... options) {
+    com.google.api.services.storage.model.HmacKeyMetadata pb = hmacKeyMetadata.toPb();
+    Map<StorageRpc.Option, ?> optionsMap = optionMap(options);
+    ExceptionHandler exceptionHandler = retryAlgorithmManager.getForHmacKeyUpdate(pb, optionsMap);
     return run(
-        () -> storageRpc.updateHmacKey(hmacKeyMetadata.toPb(), optionMap(options)),
-        HmacKeyMetadata::fromPb);
+        exceptionHandler, () -> storageRpc.updateHmacKey(pb, optionsMap), HmacKeyMetadata::fromPb);
   }
 
   @Override
@@ -1168,19 +1293,27 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
 
   @Override
   public void deleteHmacKey(final HmacKeyMetadata metadata, final DeleteHmacKeyOption... options) {
+    com.google.api.services.storage.model.HmacKeyMetadata pb = metadata.toPb();
+    Map<StorageRpc.Option, ?> optionsMap = optionMap(options);
+    ExceptionHandler exceptionHandler = retryAlgorithmManager.getForHmacKeyDelete(pb, optionsMap);
     run(
+        exceptionHandler,
         (Callable<Void>)
             () -> {
-              storageRpc.deleteHmacKey(metadata.toPb(), optionMap(options));
+              storageRpc.deleteHmacKey(pb, optionsMap);
               return null;
             },
         Function.identity());
   }
 
   private static Page<HmacKeyMetadata> listHmacKeys(
-      final StorageOptions serviceOptions, final Map<StorageRpc.Option, ?> options) {
+      final StorageOptions serviceOptions,
+      final RetryAlgorithmManager retryAlgorithmManager,
+      final Map<StorageRpc.Option, ?> options) {
+    ExceptionHandler exceptionHandler = retryAlgorithmManager.getForHmacKeyList(options);
     return Retrying.run(
         serviceOptions,
+        exceptionHandler,
         () -> serviceOptions.getStorageRpcV1().listHmacKeys(options),
         (result) -> {
           String cursor = result.x();
@@ -1189,23 +1322,33 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
                   ? ImmutableList.of()
                   : Iterables.transform(result.y(), HmacKeyMetadata::fromPb);
           return new PageImpl<>(
-              new HmacKeyMetadataPageFetcher(serviceOptions, options), cursor, metadata);
+              new HmacKeyMetadataPageFetcher(serviceOptions, retryAlgorithmManager, options),
+              cursor,
+              metadata);
         });
   }
 
   @Override
   public Policy getIamPolicy(final String bucket, BucketSourceOption... options) {
     final Map<StorageRpc.Option, ?> optionsMap = optionMap(options);
+    ExceptionHandler exceptionHandler =
+        retryAlgorithmManager.getForBucketsGetIamPolicy(bucket, optionsMap);
     return run(
-        () -> storageRpc.getIamPolicy(bucket, optionsMap), PolicyHelper::convertFromApiPolicy);
+        exceptionHandler,
+        () -> storageRpc.getIamPolicy(bucket, optionsMap),
+        PolicyHelper::convertFromApiPolicy);
   }
 
   @Override
   public Policy setIamPolicy(
       final String bucket, final Policy policy, BucketSourceOption... options) {
+    com.google.api.services.storage.model.Policy pb = convertToApiPolicy(policy);
     final Map<StorageRpc.Option, ?> optionsMap = optionMap(options);
+    ExceptionHandler exceptionHandler =
+        retryAlgorithmManager.getForBucketsSetIamPolicy(bucket, pb, optionsMap);
     return run(
-        () -> storageRpc.setIamPolicy(bucket, convertToApiPolicy(policy), optionsMap),
+        exceptionHandler,
+        () -> storageRpc.setIamPolicy(bucket, pb, optionsMap),
         PolicyHelper::convertFromApiPolicy);
   }
 
@@ -1213,7 +1356,10 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
   public List<Boolean> testIamPermissions(
       final String bucket, final List<String> permissions, BucketSourceOption... options) {
     final Map<StorageRpc.Option, ?> optionsMap = optionMap(options);
+    ExceptionHandler exceptionHandler =
+        retryAlgorithmManager.getForBucketsTestIamPermissions(bucket, permissions, optionsMap);
     return run(
+        exceptionHandler,
         () -> storageRpc.testIamPermissions(bucket, permissions, optionsMap),
         (response) -> {
           final Set<String> heldPermissions =
@@ -1230,17 +1376,23 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage {
   public Bucket lockRetentionPolicy(BucketInfo bucketInfo, BucketTargetOption... options) {
     final com.google.api.services.storage.model.Bucket bucketPb = bucketInfo.toPb();
     final Map<StorageRpc.Option, ?> optionsMap = optionMap(bucketInfo, options);
+    ExceptionHandler exceptionHandler =
+        retryAlgorithmManager.getForBucketsLockRetentionPolicy(bucketPb, optionsMap);
     return run(
-        () -> storageRpc.lockRetentionPolicy(bucketPb, optionsMap), (x) -> Bucket.fromPb(this, x));
+        exceptionHandler,
+        () -> storageRpc.lockRetentionPolicy(bucketPb, optionsMap),
+        (x) -> Bucket.fromPb(this, x));
   }
 
   @Override
   public ServiceAccount getServiceAccount(final String projectId) {
-    return run(() -> storageRpc.getServiceAccount(projectId), ServiceAccount::fromPb);
+    ExceptionHandler exceptionHandler = retryAlgorithmManager.getForServiceAccountGet(projectId);
+    return run(
+        exceptionHandler, () -> storageRpc.getServiceAccount(projectId), ServiceAccount::fromPb);
   }
 
-  private <T, U> U run(Callable<T> c, Function<T, U> f) {
-    return Retrying.run(getOptions(), c, f);
+  private <T, U> U run(ExceptionHandler exceptionHandler, Callable<T> c, Function<T, U> f) {
+    return Retrying.run(getOptions(), exceptionHandler, c, f);
   }
 
   private static <T> void addToOptionMap(
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageOptions.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageOptions.java
index 4a25e549..0cf60a78 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageOptions.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageOptions.java
@@ -25,6 +25,7 @@ import com.google.cloud.http.HttpTransportOptions;
 import com.google.cloud.storage.spi.StorageRpcFactory;
 import com.google.cloud.storage.spi.v1.HttpStorageRpc;
 import com.google.cloud.storage.spi.v1.StorageRpc;
+import com.google.common.base.MoreObjects;
 import com.google.common.collect.ImmutableSet;
 import java.util.Set;
 
@@ -36,6 +37,8 @@ public class StorageOptions extends ServiceOptions<Storage, StorageOptions> {
   private static final Set<String> SCOPES = ImmutableSet.of(GCS_SCOPE);
   private static final String DEFAULT_HOST = "https://storage.googleapis.com";
 
+  private final RetryAlgorithmManager retryAlgorithmManager;
+
   public static class DefaultStorageFactory implements StorageFactory {
 
     private static final StorageFactory INSTANCE = new DefaultStorageFactory();
@@ -58,6 +61,8 @@ public class StorageOptions extends ServiceOptions<Storage, StorageOptions> {
 
   public static class Builder extends ServiceOptions.Builder<Storage, StorageOptions, Builder> {
 
+    private RetryAlgorithmManager retryAlgorithmManager;
+
     private Builder() {}
 
     private Builder(StorageOptions options) {
@@ -73,14 +78,30 @@ public class StorageOptions extends ServiceOptions<Storage, StorageOptions> {
       return super.setTransportOptions(transportOptions);
     }
 
+    Builder setUseLegacyRetryAlgorithms() {
+      return setRetryAlgorithmManager(new LegacyRetryAlgorithmManager());
+    }
+
+    Builder setUseDefaultRetryAlgorithms() {
+      return setRetryAlgorithmManager(new NewRetryAlgorithmManager());
+    }
+
+    private Builder setRetryAlgorithmManager(RetryAlgorithmManager retryAlgorithmManager) {
+      this.retryAlgorithmManager = retryAlgorithmManager;
+      return this;
+    }
+
     @Override
     public StorageOptions build() {
-      return new StorageOptions(this);
+      return new StorageOptions(this, new StorageDefaults());
     }
   }
 
-  private StorageOptions(Builder builder) {
-    super(StorageFactory.class, StorageRpcFactory.class, builder, new StorageDefaults());
+  private StorageOptions(Builder builder, StorageDefaults serviceDefaults) {
+    super(StorageFactory.class, StorageRpcFactory.class, builder, serviceDefaults);
+    this.retryAlgorithmManager =
+        MoreObjects.firstNonNull(
+            builder.retryAlgorithmManager, serviceDefaults.getRetryAlgorithmManager());
   }
 
   private static class StorageDefaults implements ServiceDefaults<Storage, StorageOptions> {
@@ -99,6 +120,10 @@ public class StorageOptions extends ServiceOptions<Storage, StorageOptions> {
     public TransportOptions getDefaultTransportOptions() {
       return getDefaultHttpTransportOptions();
     }
+
+    public RetryAlgorithmManager getRetryAlgorithmManager() {
+      return new LegacyRetryAlgorithmManager();
+    }
   }
 
   public static HttpTransportOptions getDefaultHttpTransportOptions() {
@@ -121,6 +146,10 @@ public class StorageOptions extends ServiceOptions<Storage, StorageOptions> {
     return (StorageRpc) getRpc();
   }
 
+  protected RetryAlgorithmManager getRetryAlgorithmManager() {
+    return retryAlgorithmManager;
+  }
+
   /** Returns a default {@code StorageOptions} instance. */
   public static StorageOptions getDefaultInstance() {
     return newBuilder().build();
diff --git a/samples/install-without-bom/pom.xml b/samples/install-without-bom/pom.xml
index 1690c2f4..3fbd529a 100644
--- a/samples/install-without-bom/pom.xml
+++ b/samples/install-without-bom/pom.xml
@@ -29,7 +29,7 @@
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-storage</artifactId>
-      <version>2.1.5</version>
+      <version>2.1.6</version>
     </dependency>
     <!-- [END storage_install_without_bom] -->
 
diff --git a/samples/snapshot/pom.xml b/samples/snapshot/pom.xml
index cd1a26d9..b692e70d 100644
--- a/samples/snapshot/pom.xml
+++ b/samples/snapshot/pom.xml
@@ -28,7 +28,7 @@
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>google-cloud-storage</artifactId>
-      <version>2.1.5</version>
+      <version>2.1.6</version>
     </dependency>
 
     <dependency>


Test patch:
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/BlobTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/BlobTest.java
index 36752a4d..55534975 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/BlobTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/BlobTest.java
@@ -169,6 +169,8 @@ public class BlobTest {
   private Blob expectedBlob;
   private Storage serviceMockReturnsOptions = createMock(Storage.class);
   private StorageOptions mockOptions = createMock(StorageOptions.class);
+  private final RetryAlgorithmManager retryAlgorithmManager =
+      StorageOptions.getDefaultInstance().getRetryAlgorithmManager();
 
   @Before
   public void setUp() {
@@ -180,8 +182,10 @@ public class BlobTest {
     verify(storage);
   }
 
-  private void initializeExpectedBlob(int optionsCalls) {
-    expect(serviceMockReturnsOptions.getOptions()).andReturn(mockOptions).times(optionsCalls);
+  private void initializeExpectedBlob() {
+    expect(serviceMockReturnsOptions.getOptions()).andReturn(mockOptions).anyTimes();
+    expect(mockOptions.getRetryAlgorithmManager()).andReturn(retryAlgorithmManager).anyTimes();
+    replay(mockOptions);
     replay(serviceMockReturnsOptions);
     expectedBlob = new Blob(serviceMockReturnsOptions, new BlobInfo.BuilderImpl(BLOB_INFO));
   }
@@ -192,9 +196,9 @@ public class BlobTest {
 
   @Test
   public void testExists_True() throws Exception {
-    initializeExpectedBlob(1);
+    initializeExpectedBlob();
     Storage.BlobGetOption[] expectedOptions = {Storage.BlobGetOption.fields()};
-    expect(storage.getOptions()).andReturn(mockOptions);
+    expect(storage.getOptions()).andReturn(mockOptions).anyTimes();
     expect(storage.get(expectedBlob.getBlobId(), expectedOptions)).andReturn(expectedBlob);
     replay(storage);
     initializeBlob();
@@ -204,7 +208,7 @@ public class BlobTest {
   @Test
   public void testExists_False() throws Exception {
     Storage.BlobGetOption[] expectedOptions = {Storage.BlobGetOption.fields()};
-    expect(storage.getOptions()).andReturn(null);
+    expect(storage.getOptions()).andReturn(mockOptions).anyTimes();
     expect(storage.get(BLOB_INFO.getBlobId(), expectedOptions)).andReturn(null);
     replay(storage);
     initializeBlob();
@@ -213,9 +217,9 @@ public class BlobTest {
 
   @Test
   public void testContent() throws Exception {
-    initializeExpectedBlob(2);
+    initializeExpectedBlob();
     byte[] content = {1, 2};
-    expect(storage.getOptions()).andReturn(mockOptions);
+    expect(storage.getOptions()).andReturn(mockOptions).anyTimes();
     expect(storage.readAllBytes(BLOB_INFO.getBlobId())).andReturn(content);
     replay(storage);
     initializeBlob();
@@ -224,9 +228,9 @@ public class BlobTest {
 
   @Test
   public void testContentWithDecryptionKey() throws Exception {
-    initializeExpectedBlob(2);
+    initializeExpectedBlob();
     byte[] content = {1, 2};
-    expect(storage.getOptions()).andReturn(mockOptions);
+    expect(storage.getOptions()).andReturn(mockOptions).anyTimes();
     expect(
             storage.readAllBytes(
                 BLOB_INFO.getBlobId(), Storage.BlobSourceOption.decryptionKey(BASE64_KEY)))
@@ -240,9 +244,9 @@ public class BlobTest {
 
   @Test
   public void testReload() throws Exception {
-    initializeExpectedBlob(2);
+    initializeExpectedBlob();
     Blob expectedReloadedBlob = expectedBlob.toBuilder().setCacheControl("c").build();
-    expect(storage.getOptions()).andReturn(mockOptions);
+    expect(storage.getOptions()).andReturn(mockOptions).anyTimes();
     expect(storage.get(BLOB_INFO_NO_GENERATION.getBlobId(), new Storage.BlobGetOption[0]))
         .andReturn(expectedReloadedBlob);
     replay(storage);
@@ -253,8 +257,8 @@ public class BlobTest {
 
   @Test
   public void testReloadNull() throws Exception {
-    initializeExpectedBlob(1);
-    expect(storage.getOptions()).andReturn(mockOptions);
+    initializeExpectedBlob();
+    expect(storage.getOptions()).andReturn(mockOptions).anyTimes();
     expect(storage.get(BLOB_INFO_NO_GENERATION.getBlobId(), new Storage.BlobGetOption[0]))
         .andReturn(null);
     replay(storage);
@@ -265,10 +269,10 @@ public class BlobTest {
 
   @Test
   public void testReloadWithOptions() throws Exception {
-    initializeExpectedBlob(2);
+    initializeExpectedBlob();
     Blob expectedReloadedBlob = expectedBlob.toBuilder().setCacheControl("c").build();
     Storage.BlobGetOption[] options = {Storage.BlobGetOption.metagenerationMatch(42L)};
-    expect(storage.getOptions()).andReturn(mockOptions);
+    expect(storage.getOptions()).andReturn(mockOptions).anyTimes();
     expect(storage.get(BLOB_INFO_NO_GENERATION.getBlobId(), options))
         .andReturn(expectedReloadedBlob);
     replay(storage);
@@ -279,9 +283,9 @@ public class BlobTest {
 
   @Test
   public void testUpdate() throws Exception {
-    initializeExpectedBlob(2);
+    initializeExpectedBlob();
     Blob expectedUpdatedBlob = expectedBlob.toBuilder().setCacheControl("c").build();
-    expect(storage.getOptions()).andReturn(mockOptions).times(2);
+    expect(storage.getOptions()).andReturn(mockOptions).anyTimes();
     expect(storage.update(eq(expectedUpdatedBlob), new Storage.BlobTargetOption[0]))
         .andReturn(expectedUpdatedBlob);
     replay(storage);
@@ -293,8 +297,8 @@ public class BlobTest {
 
   @Test
   public void testDelete() throws Exception {
-    initializeExpectedBlob(2);
-    expect(storage.getOptions()).andReturn(mockOptions);
+    initializeExpectedBlob();
+    expect(storage.getOptions()).andReturn(mockOptions).anyTimes();
     expect(storage.delete(BLOB_INFO.getBlobId(), new Storage.BlobSourceOption[0])).andReturn(true);
     replay(storage);
     initializeBlob();
@@ -303,11 +307,11 @@ public class BlobTest {
 
   @Test
   public void testCopyToBucket() throws Exception {
-    initializeExpectedBlob(2);
+    initializeExpectedBlob();
     BlobInfo target = BlobInfo.newBuilder(BlobId.of("bt", "n")).build();
     CopyWriter copyWriter = createMock(CopyWriter.class);
     Capture<CopyRequest> capturedCopyRequest = Capture.newInstance();
-    expect(storage.getOptions()).andReturn(mockOptions);
+    expect(storage.getOptions()).andReturn(mockOptions).anyTimes();
     expect(storage.copy(capture(capturedCopyRequest))).andReturn(copyWriter);
     replay(storage);
     initializeBlob();
@@ -322,11 +326,11 @@ public class BlobTest {
 
   @Test
   public void testCopyTo() throws Exception {
-    initializeExpectedBlob(2);
+    initializeExpectedBlob();
     BlobInfo target = BlobInfo.newBuilder(BlobId.of("bt", "nt")).build();
     CopyWriter copyWriter = createMock(CopyWriter.class);
     Capture<CopyRequest> capturedCopyRequest = Capture.newInstance();
-    expect(storage.getOptions()).andReturn(mockOptions);
+    expect(storage.getOptions()).andReturn(mockOptions).anyTimes();
     expect(storage.copy(capture(capturedCopyRequest))).andReturn(copyWriter);
     replay(storage);
     initializeBlob();
@@ -341,12 +345,12 @@ public class BlobTest {
 
   @Test
   public void testCopyToBlobId() throws Exception {
-    initializeExpectedBlob(2);
+    initializeExpectedBlob();
     BlobInfo target = BlobInfo.newBuilder(BlobId.of("bt", "nt")).build();
     BlobId targetId = BlobId.of("bt", "nt");
     CopyWriter copyWriter = createMock(CopyWriter.class);
     Capture<CopyRequest> capturedCopyRequest = Capture.newInstance();
-    expect(storage.getOptions()).andReturn(mockOptions);
+    expect(storage.getOptions()).andReturn(mockOptions).anyTimes();
     expect(storage.copy(capture(capturedCopyRequest))).andReturn(copyWriter);
     replay(storage);
     initializeBlob();
@@ -361,9 +365,9 @@ public class BlobTest {
 
   @Test
   public void testReader() throws Exception {
-    initializeExpectedBlob(2);
+    initializeExpectedBlob();
     ReadChannel channel = createMock(ReadChannel.class);
-    expect(storage.getOptions()).andReturn(mockOptions);
+    expect(storage.getOptions()).andReturn(mockOptions).anyTimes();
     expect(storage.reader(BLOB_INFO.getBlobId())).andReturn(channel);
     replay(storage);
     initializeBlob();
@@ -372,9 +376,9 @@ public class BlobTest {
 
   @Test
   public void testReaderWithDecryptionKey() throws Exception {
-    initializeExpectedBlob(2);
+    initializeExpectedBlob();
     ReadChannel channel = createMock(ReadChannel.class);
-    expect(storage.getOptions()).andReturn(mockOptions);
+    expect(storage.getOptions()).andReturn(mockOptions).anyTimes();
     expect(
             storage.reader(
                 BLOB_INFO.getBlobId(), Storage.BlobSourceOption.decryptionKey(BASE64_KEY)))
@@ -388,9 +392,9 @@ public class BlobTest {
 
   @Test
   public void testWriter() throws Exception {
-    initializeExpectedBlob(2);
+    initializeExpectedBlob();
     BlobWriteChannel channel = createMock(BlobWriteChannel.class);
-    expect(storage.getOptions()).andReturn(mockOptions);
+    expect(storage.getOptions()).andReturn(mockOptions).anyTimes();
     expect(storage.writer(eq(expectedBlob))).andReturn(channel);
     replay(storage);
     initializeBlob();
@@ -399,9 +403,9 @@ public class BlobTest {
 
   @Test
   public void testWriterWithEncryptionKey() throws Exception {
-    initializeExpectedBlob(2);
+    initializeExpectedBlob();
     BlobWriteChannel channel = createMock(BlobWriteChannel.class);
-    expect(storage.getOptions()).andReturn(mockOptions);
+    expect(storage.getOptions()).andReturn(mockOptions).anyTimes();
     expect(storage.writer(eq(expectedBlob), eq(BlobWriteOption.encryptionKey(BASE64_KEY))))
         .andReturn(channel)
         .times(2);
@@ -413,9 +417,9 @@ public class BlobTest {
 
   @Test
   public void testWriterWithKmsKeyName() throws Exception {
-    initializeExpectedBlob(2);
+    initializeExpectedBlob();
     BlobWriteChannel channel = createMock(BlobWriteChannel.class);
-    expect(storage.getOptions()).andReturn(mockOptions);
+    expect(storage.getOptions()).andReturn(mockOptions).anyTimes();
     expect(storage.writer(eq(expectedBlob), eq(BlobWriteOption.kmsKeyName(KMS_KEY_NAME))))
         .andReturn(channel);
     replay(storage);
@@ -425,9 +429,9 @@ public class BlobTest {
 
   @Test
   public void testSignUrl() throws Exception {
-    initializeExpectedBlob(2);
+    initializeExpectedBlob();
     URL url = new URL("http://localhost:123/bla");
-    expect(storage.getOptions()).andReturn(mockOptions);
+    expect(storage.getOptions()).andReturn(mockOptions).anyTimes();
     expect(storage.signUrl(expectedBlob, 100, TimeUnit.SECONDS)).andReturn(url);
     replay(storage);
     initializeBlob();
@@ -436,8 +440,8 @@ public class BlobTest {
 
   @Test
   public void testGetAcl() throws Exception {
-    initializeExpectedBlob(1);
-    expect(storage.getOptions()).andReturn(mockOptions);
+    initializeExpectedBlob();
+    expect(storage.getOptions()).andReturn(mockOptions).anyTimes();
     expect(storage.getAcl(BLOB_INFO.getBlobId(), User.ofAllAuthenticatedUsers())).andReturn(ACL);
     replay(storage);
     initializeBlob();
@@ -446,8 +450,8 @@ public class BlobTest {
 
   @Test
   public void testDeleteAcl() throws Exception {
-    initializeExpectedBlob(1);
-    expect(storage.getOptions()).andReturn(mockOptions);
+    initializeExpectedBlob();
+    expect(storage.getOptions()).andReturn(mockOptions).anyTimes();
     expect(storage.deleteAcl(BLOB_INFO.getBlobId(), User.ofAllAuthenticatedUsers()))
         .andReturn(true);
     replay(storage);
@@ -457,8 +461,8 @@ public class BlobTest {
 
   @Test
   public void testCreateAcl() throws Exception {
-    initializeExpectedBlob(1);
-    expect(storage.getOptions()).andReturn(mockOptions);
+    initializeExpectedBlob();
+    expect(storage.getOptions()).andReturn(mockOptions).anyTimes();
     Acl returnedAcl = ACL.toBuilder().setEtag("ETAG").setId("ID").build();
     expect(storage.createAcl(BLOB_INFO.getBlobId(), ACL)).andReturn(returnedAcl);
     replay(storage);
@@ -468,8 +472,8 @@ public class BlobTest {
 
   @Test
   public void testUpdateAcl() throws Exception {
-    initializeExpectedBlob(1);
-    expect(storage.getOptions()).andReturn(mockOptions);
+    initializeExpectedBlob();
+    expect(storage.getOptions()).andReturn(mockOptions).anyTimes();
     Acl returnedAcl = ACL.toBuilder().setEtag("ETAG").setId("ID").build();
     expect(storage.updateAcl(BLOB_INFO.getBlobId(), ACL)).andReturn(returnedAcl);
     replay(storage);
@@ -479,8 +483,8 @@ public class BlobTest {
 
   @Test
   public void testListAcls() throws Exception {
-    initializeExpectedBlob(1);
-    expect(storage.getOptions()).andReturn(mockOptions);
+    initializeExpectedBlob();
+    expect(storage.getOptions()).andReturn(mockOptions).anyTimes();
     expect(storage.listAcls(BLOB_INFO.getBlobId())).andReturn(ACLS);
     replay(storage);
     initializeBlob();
@@ -489,8 +493,10 @@ public class BlobTest {
 
   @Test
   public void testToBuilder() {
-    expect(storage.getOptions()).andReturn(mockOptions).times(6);
+    expect(storage.getOptions()).andReturn(mockOptions).anyTimes();
+    expect(mockOptions.getRetryAlgorithmManager()).andReturn(retryAlgorithmManager).anyTimes();
     replay(storage);
+    replay(mockOptions);
     Blob fullBlob = new Blob(storage, new BlobInfo.BuilderImpl(FULL_BLOB_INFO));
     assertEquals(fullBlob, fullBlob.toBuilder().build());
     Blob simpleBlob = new Blob(storage, new BlobInfo.BuilderImpl(BLOB_INFO));
@@ -501,8 +507,8 @@ public class BlobTest {
 
   @Test
   public void testBuilder() {
-    initializeExpectedBlob(4);
-    expect(storage.getOptions()).andReturn(mockOptions).times(6);
+    initializeExpectedBlob();
+    expect(storage.getOptions()).andReturn(mockOptions).anyTimes();
     replay(storage);
     Blob.Builder builder = new Blob.Builder(new Blob(storage, new BlobInfo.BuilderImpl(BLOB_INFO)));
     Blob blob =
@@ -609,11 +615,12 @@ public class BlobTest {
 
   private StorageRpc prepareForDownload() {
     StorageRpc mockStorageRpc = createNiceMock(StorageRpc.class);
-    expect(storage.getOptions()).andReturn(mockOptions);
+    expect(storage.getOptions()).andReturn(mockOptions).anyTimes();
     replay(storage);
     expect(mockOptions.getStorageRpcV1()).andReturn(mockStorageRpc);
     expect(mockOptions.getRetrySettings()).andReturn(RETRY_SETTINGS);
     expect(mockOptions.getClock()).andReturn(API_CLOCK);
+    expect(mockOptions.getRetryAlgorithmManager()).andReturn(retryAlgorithmManager).anyTimes();
     replay(mockOptions);
     blob = new Blob(storage, new BlobInfo.BuilderImpl(BLOB_INFO));
     return mockStorageRpc;
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/BlobWriteChannelTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/BlobWriteChannelTest.java
index d25bd64a..d9a70897 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/BlobWriteChannelTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/BlobWriteChannelTest.java
@@ -35,6 +35,8 @@ import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
 import com.google.api.services.storage.model.StorageObject;
+import com.google.cloud.ExceptionHandler;
+import com.google.cloud.NoCredentials;
 import com.google.cloud.RestorableState;
 import com.google.cloud.WriteChannel;
 import com.google.cloud.storage.spi.StorageRpcFactory;
@@ -47,6 +49,7 @@ import java.net.SocketException;
 import java.net.URL;
 import java.nio.ByteBuffer;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.Map;
 import java.util.Random;
 import org.easymock.Capture;
@@ -76,6 +79,7 @@ public class BlobWriteChannelTest {
   private StorageRpcFactory rpcFactoryMock;
   private StorageRpc storageRpcMock;
   private BlobWriteChannel writer;
+  private RetryAlgorithmManager retryAlgorithmManager;
 
   @Before
   public void setUp() {
@@ -87,7 +91,9 @@ public class BlobWriteChannelTest {
         StorageOptions.newBuilder()
             .setProjectId("projectid")
             .setServiceRpcFactory(rpcFactoryMock)
+            .setCredentials(NoCredentials.getInstance())
             .build();
+    retryAlgorithmManager = options.getRetryAlgorithmManager();
   }
 
   @After
@@ -99,7 +105,7 @@ public class BlobWriteChannelTest {
   public void testCreate() {
     expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);
     replay(storageRpcMock);
-    writer = new BlobWriteChannel(options, BLOB_INFO, EMPTY_RPC_OPTIONS);
+    writer = newWriter();
     assertTrue(writer.isOpen());
     assertNull(writer.getStorageObject());
   }
@@ -110,7 +116,7 @@ public class BlobWriteChannelTest {
         .andThrow(socketClosedException);
     expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);
     replay(storageRpcMock);
-    writer = new BlobWriteChannel(options, BLOB_INFO, EMPTY_RPC_OPTIONS);
+    writer = newWriter();
     assertTrue(writer.isOpen());
     assertNull(writer.getStorageObject());
   }
@@ -121,7 +127,7 @@ public class BlobWriteChannelTest {
         .andThrow(new RuntimeException());
     replay(storageRpcMock);
     try {
-      new BlobWriteChannel(options, BLOB_INFO, EMPTY_RPC_OPTIONS);
+      newWriter();
       Assert.fail();
     } catch (RuntimeException ex) {
       assertNotNull(ex.getMessage());
@@ -129,15 +135,15 @@ public class BlobWriteChannelTest {
   }
 
   @Test
-  public void testWriteWithoutFlush() throws IOException {
+  public void testWriteWithoutFlush() throws Exception {
     expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);
     replay(storageRpcMock);
-    writer = new BlobWriteChannel(options, BLOB_INFO, EMPTY_RPC_OPTIONS);
+    writer = newWriter();
     assertEquals(MIN_CHUNK_SIZE, writer.write(ByteBuffer.allocate(MIN_CHUNK_SIZE)));
   }
 
   @Test
-  public void testWriteWithFlushRetryChunk() throws IOException {
+  public void testWriteWithFlushRetryChunk() throws Exception {
     ByteBuffer buffer = randomBuffer(MIN_CHUNK_SIZE);
     Capture<byte[]> capturedBuffer = Capture.newInstance();
     expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);
@@ -161,7 +167,7 @@ public class BlobWriteChannelTest {
                 eq(false)))
         .andReturn(null);
     replay(storageRpcMock);
-    writer = new BlobWriteChannel(options, BLOB_INFO, EMPTY_RPC_OPTIONS);
+    writer = newWriter();
     writer.setChunkSize(MIN_CHUNK_SIZE);
     assertEquals(MIN_CHUNK_SIZE, writer.write(buffer));
     assertTrue(writer.isOpen());
@@ -170,7 +176,7 @@ public class BlobWriteChannelTest {
   }
 
   @Test
-  public void testWriteWithRetryFullChunk() throws IOException {
+  public void testWriteWithRetryFullChunk() throws Exception {
     ByteBuffer buffer = randomBuffer(MIN_CHUNK_SIZE);
     Capture<byte[]> capturedBuffer = Capture.newInstance();
     expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);
@@ -198,7 +204,7 @@ public class BlobWriteChannelTest {
                 eq(true)))
         .andReturn(BLOB_INFO.toPb());
     replay(storageRpcMock);
-    writer = new BlobWriteChannel(options, BLOB_INFO, EMPTY_RPC_OPTIONS);
+    writer = newWriter();
     writer.setChunkSize(MIN_CHUNK_SIZE);
     assertEquals(MIN_CHUNK_SIZE, writer.write(buffer));
     writer.close();
@@ -208,7 +214,7 @@ public class BlobWriteChannelTest {
   }
 
   @Test
-  public void testWriteWithRemoteProgressMade() throws IOException {
+  public void testWriteWithRemoteProgressMade() throws Exception {
     ByteBuffer buffer = randomBuffer(MIN_CHUNK_SIZE);
     Capture<byte[]> capturedBuffer = Capture.newInstance();
     expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);
@@ -233,7 +239,7 @@ public class BlobWriteChannelTest {
                 eq(false)))
         .andReturn(null);
     replay(storageRpcMock);
-    writer = new BlobWriteChannel(options, BLOB_INFO, EMPTY_RPC_OPTIONS);
+    writer = newWriter();
     writer.setChunkSize(MIN_CHUNK_SIZE);
     assertEquals(MIN_CHUNK_SIZE, writer.write(buffer));
     assertTrue(writer.isOpen());
@@ -242,7 +248,7 @@ public class BlobWriteChannelTest {
   }
 
   @Test
-  public void testWriteWithDriftRetryCase4() throws IOException {
+  public void testWriteWithDriftRetryCase4() throws Exception {
     ByteBuffer buffer = randomBuffer(MIN_CHUNK_SIZE);
     Capture<byte[]> capturedBuffer = Capture.newInstance();
     expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);
@@ -266,7 +272,7 @@ public class BlobWriteChannelTest {
                 eq(false)))
         .andReturn(null);
     replay(storageRpcMock);
-    writer = new BlobWriteChannel(options, BLOB_INFO, EMPTY_RPC_OPTIONS);
+    writer = newWriter();
     writer.setChunkSize(MIN_CHUNK_SIZE);
     assertEquals(MIN_CHUNK_SIZE, writer.write(buffer));
     assertArrayEquals(buffer.array(), capturedBuffer.getValue());
@@ -279,7 +285,7 @@ public class BlobWriteChannelTest {
   }
 
   @Test
-  public void testWriteWithUnreachableRemoteOffset() throws IOException {
+  public void testWriteWithUnreachableRemoteOffset() throws Exception {
     ByteBuffer buffer = randomBuffer(MIN_CHUNK_SIZE);
     Capture<byte[]> capturedBuffer = Capture.newInstance();
     expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);
@@ -294,7 +300,7 @@ public class BlobWriteChannelTest {
         .andThrow(socketClosedException);
     expect(storageRpcMock.getCurrentUploadOffset(eq(UPLOAD_ID))).andReturn(MIN_CHUNK_SIZE + 10L);
     replay(storageRpcMock);
-    writer = new BlobWriteChannel(options, BLOB_INFO, EMPTY_RPC_OPTIONS);
+    writer = newWriter();
     writer.setChunkSize(MIN_CHUNK_SIZE);
     try {
       writer.write(buffer);
@@ -308,7 +314,7 @@ public class BlobWriteChannelTest {
   }
 
   @Test
-  public void testWriteWithRetryAndObjectMetadata() throws IOException {
+  public void testWriteWithRetryAndObjectMetadata() throws Exception {
     ByteBuffer buffer = randomBuffer(MIN_CHUNK_SIZE);
     Capture<byte[]> capturedBuffer = Capture.newInstance();
     expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);
@@ -339,7 +345,7 @@ public class BlobWriteChannelTest {
     expect(storageRpcMock.queryCompletedResumableUpload(eq(UPLOAD_ID), eq((long) MIN_CHUNK_SIZE)))
         .andReturn(BLOB_INFO.toPb().setSize(BigInteger.valueOf(MIN_CHUNK_SIZE)));
     replay(storageRpcMock);
-    writer = new BlobWriteChannel(options, BLOB_INFO, EMPTY_RPC_OPTIONS);
+    writer = newWriter();
     writer.setChunkSize(MIN_CHUNK_SIZE);
     assertEquals(MIN_CHUNK_SIZE, writer.write(buffer));
     writer.close();
@@ -349,7 +355,7 @@ public class BlobWriteChannelTest {
   }
 
   @Test
-  public void testWriteWithUploadCompletedByAnotherClient() throws IOException {
+  public void testWriteWithUploadCompletedByAnotherClient() throws Exception {
     ByteBuffer buffer = randomBuffer(MIN_CHUNK_SIZE);
     Capture<byte[]> capturedBuffer = Capture.newInstance();
     expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);
@@ -374,7 +380,7 @@ public class BlobWriteChannelTest {
     expect(storageRpcMock.getCurrentUploadOffset(eq(UPLOAD_ID))).andReturn(-1L);
     expect(storageRpcMock.getCurrentUploadOffset(eq(UPLOAD_ID))).andReturn(-1L);
     replay(storageRpcMock);
-    writer = new BlobWriteChannel(options, BLOB_INFO, EMPTY_RPC_OPTIONS);
+    writer = newWriter();
     writer.setChunkSize(MIN_CHUNK_SIZE);
     try {
       writer.write(buffer);
@@ -390,7 +396,7 @@ public class BlobWriteChannelTest {
   }
 
   @Test
-  public void testWriteWithLocalOffsetGoingBeyondRemoteOffset() throws IOException {
+  public void testWriteWithLocalOffsetGoingBeyondRemoteOffset() throws Exception {
     ByteBuffer buffer = randomBuffer(MIN_CHUNK_SIZE);
     Capture<byte[]> capturedBuffer = Capture.newInstance();
     expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);
@@ -414,7 +420,7 @@ public class BlobWriteChannelTest {
         .andThrow(socketClosedException);
     expect(storageRpcMock.getCurrentUploadOffset(eq(UPLOAD_ID))).andReturn(0L);
     replay(storageRpcMock);
-    writer = new BlobWriteChannel(options, BLOB_INFO, EMPTY_RPC_OPTIONS);
+    writer = newWriter();
     writer.setChunkSize(MIN_CHUNK_SIZE);
     try {
       writer.write(buffer);
@@ -428,7 +434,7 @@ public class BlobWriteChannelTest {
   }
 
   @Test
-  public void testGetCurrentUploadOffset() throws IOException {
+  public void testGetCurrentUploadOffset() throws Exception {
     ByteBuffer buffer = randomBuffer(MIN_CHUNK_SIZE);
     Capture<byte[]> capturedBuffer = Capture.newInstance();
     expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);
@@ -462,7 +468,7 @@ public class BlobWriteChannelTest {
                 eq(true)))
         .andReturn(BLOB_INFO.toPb());
     replay(storageRpcMock);
-    writer = new BlobWriteChannel(options, BLOB_INFO, EMPTY_RPC_OPTIONS);
+    writer = newWriter();
     writer.setChunkSize(MIN_CHUNK_SIZE);
     assertEquals(MIN_CHUNK_SIZE, writer.write(buffer));
     writer.close();
@@ -472,7 +478,7 @@ public class BlobWriteChannelTest {
   }
 
   @Test
-  public void testWriteWithLastFlushRetryChunkButCompleted() throws IOException {
+  public void testWriteWithLastFlushRetryChunkButCompleted() throws Exception {
     ByteBuffer buffer = randomBuffer(MIN_CHUNK_SIZE);
     Capture<byte[]> capturedBuffer = Capture.newInstance();
     expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);
@@ -489,7 +495,7 @@ public class BlobWriteChannelTest {
     expect(storageRpcMock.queryCompletedResumableUpload(eq(UPLOAD_ID), eq((long) MIN_CHUNK_SIZE)))
         .andReturn(BLOB_INFO.toPb().setSize(BigInteger.valueOf(MIN_CHUNK_SIZE)));
     replay(storageRpcMock);
-    writer = new BlobWriteChannel(options, BLOB_INFO, EMPTY_RPC_OPTIONS);
+    writer = newWriter();
     assertEquals(MIN_CHUNK_SIZE, writer.write(buffer));
     writer.close();
     assertFalse(writer.isRetrying());
@@ -501,7 +507,7 @@ public class BlobWriteChannelTest {
   }
 
   @Test
-  public void testWriteWithFlush() throws IOException {
+  public void testWriteWithFlush() throws Exception {
     expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);
     Capture<byte[]> capturedBuffer = Capture.newInstance();
     expect(
@@ -514,7 +520,7 @@ public class BlobWriteChannelTest {
                 eq(false)))
         .andReturn(null);
     replay(storageRpcMock);
-    writer = new BlobWriteChannel(options, BLOB_INFO, EMPTY_RPC_OPTIONS);
+    writer = newWriter();
     writer.setChunkSize(CUSTOM_CHUNK_SIZE);
     ByteBuffer buffer = randomBuffer(CUSTOM_CHUNK_SIZE);
     assertEquals(CUSTOM_CHUNK_SIZE, writer.write(buffer));
@@ -523,7 +529,7 @@ public class BlobWriteChannelTest {
   }
 
   @Test
-  public void testWritesAndFlush() throws IOException {
+  public void testWritesAndFlush() throws Exception {
     expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);
     Capture<byte[]> capturedBuffer = Capture.newInstance();
     expect(
@@ -536,7 +542,7 @@ public class BlobWriteChannelTest {
                 eq(false)))
         .andReturn(null);
     replay(storageRpcMock);
-    writer = new BlobWriteChannel(options, BLOB_INFO, EMPTY_RPC_OPTIONS);
+    writer = newWriter();
     ByteBuffer[] buffers = new ByteBuffer[DEFAULT_CHUNK_SIZE / MIN_CHUNK_SIZE];
     for (int i = 0; i < buffers.length; i++) {
       buffers[i] = randomBuffer(MIN_CHUNK_SIZE);
@@ -552,7 +558,7 @@ public class BlobWriteChannelTest {
   }
 
   @Test
-  public void testCloseWithoutFlush() throws IOException {
+  public void testCloseWithoutFlush() throws Exception {
     expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);
     Capture<byte[]> capturedBuffer = Capture.newInstance();
     expect(
@@ -560,7 +566,7 @@ public class BlobWriteChannelTest {
                 eq(UPLOAD_ID), capture(capturedBuffer), eq(0), eq(0L), eq(0), eq(true)))
         .andReturn(UPDATED_BLOB);
     replay(storageRpcMock);
-    writer = new BlobWriteChannel(options, BLOB_INFO, EMPTY_RPC_OPTIONS);
+    writer = newWriter();
     assertTrue(writer.isOpen());
     writer.close();
     assertArrayEquals(new byte[0], capturedBuffer.getValue());
@@ -569,7 +575,7 @@ public class BlobWriteChannelTest {
   }
 
   @Test
-  public void testCloseWithFlush() throws IOException {
+  public void testCloseWithFlush() throws Exception {
     expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);
     Capture<byte[]> capturedBuffer = Capture.newInstance();
     ByteBuffer buffer = randomBuffer(MIN_CHUNK_SIZE);
@@ -583,7 +589,7 @@ public class BlobWriteChannelTest {
                 eq(true)))
         .andReturn(UPDATED_BLOB);
     replay(storageRpcMock);
-    writer = new BlobWriteChannel(options, BLOB_INFO, EMPTY_RPC_OPTIONS);
+    writer = newWriter();
     assertTrue(writer.isOpen());
     writer.write(buffer);
     writer.close();
@@ -594,7 +600,7 @@ public class BlobWriteChannelTest {
   }
 
   @Test
-  public void testWriteClosed() throws IOException {
+  public void testWriteClosed() throws Exception {
     expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);
     Capture<byte[]> capturedBuffer = Capture.newInstance();
     expect(
@@ -602,7 +608,7 @@ public class BlobWriteChannelTest {
                 eq(UPLOAD_ID), capture(capturedBuffer), eq(0), eq(0L), eq(0), eq(true)))
         .andReturn(UPDATED_BLOB);
     replay(storageRpcMock);
-    writer = new BlobWriteChannel(options, BLOB_INFO, EMPTY_RPC_OPTIONS);
+    writer = newWriter();
     writer.close();
     try {
       writer.write(ByteBuffer.allocate(MIN_CHUNK_SIZE));
@@ -614,7 +620,7 @@ public class BlobWriteChannelTest {
   }
 
   @Test
-  public void testSaveAndRestore() throws IOException {
+  public void testSaveAndRestore() throws Exception {
     expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);
     Capture<byte[]> capturedBuffer = Capture.newInstance(CaptureType.ALL);
     Capture<Long> capturedPosition = Capture.newInstance(CaptureType.ALL);
@@ -631,7 +637,7 @@ public class BlobWriteChannelTest {
     replay(storageRpcMock);
     ByteBuffer buffer1 = randomBuffer(DEFAULT_CHUNK_SIZE);
     ByteBuffer buffer2 = randomBuffer(DEFAULT_CHUNK_SIZE);
-    writer = new BlobWriteChannel(options, BLOB_INFO, EMPTY_RPC_OPTIONS);
+    writer = newWriter();
     assertEquals(DEFAULT_CHUNK_SIZE, writer.write(buffer1));
     assertArrayEquals(buffer1.array(), capturedBuffer.getValues().get(0));
     assertEquals(new Long(0L), capturedPosition.getValues().get(0));
@@ -643,7 +649,7 @@ public class BlobWriteChannelTest {
   }
 
   @Test
-  public void testSaveAndRestoreClosed() throws IOException {
+  public void testSaveAndRestoreClosed() throws Exception {
     expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID);
     Capture<byte[]> capturedBuffer = Capture.newInstance();
     expect(
@@ -651,7 +657,7 @@ public class BlobWriteChannelTest {
                 eq(UPLOAD_ID), capture(capturedBuffer), eq(0), eq(0L), eq(0), eq(true)))
         .andReturn(UPDATED_BLOB);
     replay(storageRpcMock);
-    writer = new BlobWriteChannel(options, BLOB_INFO, EMPTY_RPC_OPTIONS);
+    writer = newWriter();
     writer.close();
     RestorableState<WriteChannel> writerState = writer.capture();
     RestorableState<WriteChannel> expectedWriterState =
@@ -670,10 +676,10 @@ public class BlobWriteChannelTest {
   public void testStateEquals() {
     expect(storageRpcMock.open(BLOB_INFO.toPb(), EMPTY_RPC_OPTIONS)).andReturn(UPLOAD_ID).times(2);
     replay(storageRpcMock);
-    writer = new BlobWriteChannel(options, BLOB_INFO, EMPTY_RPC_OPTIONS);
+    writer = newWriter();
     // avoid closing when you don't want partial writes to GCS upon failure
     @SuppressWarnings("resource")
-    WriteChannel writer2 = new BlobWriteChannel(options, BLOB_INFO, EMPTY_RPC_OPTIONS);
+    WriteChannel writer2 = newWriter();
     RestorableState<WriteChannel> state = writer.capture();
     RestorableState<WriteChannel> state2 = writer2.capture();
     assertEquals(state, state2);
@@ -682,15 +688,7 @@ public class BlobWriteChannelTest {
   }
 
   @Test
-  public void testWriteWithSignedURLAndWithoutFlush() throws IOException {
-    expect(storageRpcMock.open(SIGNED_URL)).andReturn(UPLOAD_ID);
-    replay(storageRpcMock);
-    writer = new BlobWriteChannel(options, new URL(SIGNED_URL));
-    assertEquals(MIN_CHUNK_SIZE, writer.write(ByteBuffer.allocate(MIN_CHUNK_SIZE)));
-  }
-
-  @Test
-  public void testWriteWithSignedURLAndWithFlush() throws IOException {
+  public void testWriteWithSignedURLAndWithFlush() throws Exception {
     expect(storageRpcMock.open(SIGNED_URL)).andReturn(UPLOAD_ID);
     Capture<byte[]> capturedBuffer = Capture.newInstance();
     expect(
@@ -703,7 +701,7 @@ public class BlobWriteChannelTest {
                 eq(false)))
         .andReturn(null);
     replay(storageRpcMock);
-    writer = new BlobWriteChannel(options, new URL(SIGNED_URL));
+    writer = newWriterForSignedUrl();
     writer.setChunkSize(CUSTOM_CHUNK_SIZE);
     ByteBuffer buffer = randomBuffer(CUSTOM_CHUNK_SIZE);
     assertEquals(CUSTOM_CHUNK_SIZE, writer.write(buffer));
@@ -711,7 +709,7 @@ public class BlobWriteChannelTest {
   }
 
   @Test
-  public void testWriteWithSignedURLAndFlush() throws IOException {
+  public void testWriteWithSignedURLAndFlush() throws Exception {
     expect(storageRpcMock.open(SIGNED_URL)).andReturn(UPLOAD_ID);
     Capture<byte[]> capturedBuffer = Capture.newInstance();
     expect(
@@ -724,7 +722,7 @@ public class BlobWriteChannelTest {
                 eq(false)))
         .andReturn(null);
     replay(storageRpcMock);
-    writer = new BlobWriteChannel(options, new URL(SIGNED_URL));
+    writer = newWriterForSignedUrl();
     ByteBuffer[] buffers = new ByteBuffer[DEFAULT_CHUNK_SIZE / MIN_CHUNK_SIZE];
     for (int i = 0; i < buffers.length; i++) {
       buffers[i] = randomBuffer(MIN_CHUNK_SIZE);
@@ -739,7 +737,7 @@ public class BlobWriteChannelTest {
   }
 
   @Test
-  public void testCloseWithSignedURLWithoutFlush() throws IOException {
+  public void testCloseWithSignedURLWithoutFlush() throws Exception {
     expect(storageRpcMock.open(SIGNED_URL)).andReturn(UPLOAD_ID);
     Capture<byte[]> capturedBuffer = Capture.newInstance();
     expect(
@@ -747,7 +745,7 @@ public class BlobWriteChannelTest {
                 eq(UPLOAD_ID), capture(capturedBuffer), eq(0), eq(0L), eq(0), eq(true)))
         .andReturn(UPDATED_BLOB);
     replay(storageRpcMock);
-    writer = new BlobWriteChannel(options, new URL(SIGNED_URL));
+    writer = newWriterForSignedUrl();
     assertTrue(writer.isOpen());
     writer.close();
     assertArrayEquals(new byte[0], capturedBuffer.getValue());
@@ -755,7 +753,7 @@ public class BlobWriteChannelTest {
   }
 
   @Test
-  public void testCloseWithSignedURLWithFlush() throws IOException {
+  public void testCloseWithSignedURLWithFlush() throws Exception {
     expect(storageRpcMock.open(SIGNED_URL)).andReturn(UPLOAD_ID);
     Capture<byte[]> capturedBuffer = Capture.newInstance();
     ByteBuffer buffer = randomBuffer(MIN_CHUNK_SIZE);
@@ -769,7 +767,7 @@ public class BlobWriteChannelTest {
                 eq(true)))
         .andReturn(UPDATED_BLOB);
     replay(storageRpcMock);
-    writer = new BlobWriteChannel(options, new URL(SIGNED_URL));
+    writer = newWriterForSignedUrl();
     assertTrue(writer.isOpen());
     writer.write(buffer);
     writer.close();
@@ -779,7 +777,7 @@ public class BlobWriteChannelTest {
   }
 
   @Test
-  public void testWriteWithSignedURLClosed() throws IOException {
+  public void testWriteWithSignedURLClosed() throws Exception {
     expect(storageRpcMock.open(SIGNED_URL)).andReturn(UPLOAD_ID);
     Capture<byte[]> capturedBuffer = Capture.newInstance();
     expect(
@@ -787,7 +785,7 @@ public class BlobWriteChannelTest {
                 eq(UPLOAD_ID), capture(capturedBuffer), eq(0), eq(0L), eq(0), eq(true)))
         .andReturn(UPDATED_BLOB);
     replay(storageRpcMock);
-    writer = new BlobWriteChannel(options, new URL(SIGNED_URL));
+    writer = newWriterForSignedUrl();
     writer.close();
     try {
       writer.write(ByteBuffer.allocate(MIN_CHUNK_SIZE));
@@ -798,7 +796,7 @@ public class BlobWriteChannelTest {
   }
 
   @Test
-  public void testSaveAndRestoreWithSignedURL() throws IOException {
+  public void testSaveAndRestoreWithSignedURL() throws Exception {
     expect(storageRpcMock.open(SIGNED_URL)).andReturn(UPLOAD_ID);
     Capture<byte[]> capturedBuffer = Capture.newInstance(CaptureType.ALL);
     Capture<Long> capturedPosition = Capture.newInstance(CaptureType.ALL);
@@ -815,7 +813,7 @@ public class BlobWriteChannelTest {
     replay(storageRpcMock);
     ByteBuffer buffer1 = randomBuffer(DEFAULT_CHUNK_SIZE);
     ByteBuffer buffer2 = randomBuffer(DEFAULT_CHUNK_SIZE);
-    writer = new BlobWriteChannel(options, new URL(SIGNED_URL));
+    writer = newWriterForSignedUrl();
     assertEquals(DEFAULT_CHUNK_SIZE, writer.write(buffer1));
     assertArrayEquals(buffer1.array(), capturedBuffer.getValues().get(0));
     assertEquals(new Long(0L), capturedPosition.getValues().get(0));
@@ -826,18 +824,35 @@ public class BlobWriteChannelTest {
     assertEquals(new Long(DEFAULT_CHUNK_SIZE), capturedPosition.getValues().get(1));
   }
 
-  @Test
-  public void testRuntimeExceptionWithSignedURL() throws MalformedURLException {
-    String exceptionMessage = "invalid signedURL";
-    expect(new BlobWriteChannel(options, new URL(SIGNED_URL)))
-        .andThrow(new RuntimeException(exceptionMessage));
-    replay(storageRpcMock);
-    try {
-      writer = new BlobWriteChannel(options, new URL(SIGNED_URL));
-      Assert.fail();
-    } catch (StorageException ex) {
-      assertNotNull(ex.getMessage());
-    }
+  private BlobWriteChannel newWriter() {
+    Map<StorageRpc.Option, ?> optionsMap = EMPTY_RPC_OPTIONS;
+    ExceptionHandler createResultExceptionHandler =
+        retryAlgorithmManager.getForResumableUploadSessionCreate(optionsMap);
+    ExceptionHandler writeResultExceptionHandler =
+        retryAlgorithmManager.getForResumableUploadSessionWrite(optionsMap);
+    return BlobWriteChannel.newBuilder()
+        .setStorageOptions(options)
+        .setBlobInfo(BLOB_INFO)
+        .setUploadIdSupplier(
+            ResumableMedia.startUploadForBlobInfo(
+                options, BLOB_INFO, optionsMap, createResultExceptionHandler))
+        .setPutExceptionHandler(writeResultExceptionHandler)
+        .build();
+  }
+
+  private BlobWriteChannel newWriterForSignedUrl() throws MalformedURLException {
+    Map<StorageRpc.Option, Object> optionsMap = Collections.emptyMap();
+    ExceptionHandler createResultExceptionHandler =
+        retryAlgorithmManager.getForResumableUploadSessionCreate(optionsMap);
+    ExceptionHandler writeResultExceptionHandler =
+        retryAlgorithmManager.getForResumableUploadSessionWrite(optionsMap);
+    return BlobWriteChannel.newBuilder()
+        .setStorageOptions(options)
+        .setUploadIdSupplier(
+            ResumableMedia.startUploadForSignedUrl(
+                options, new URL(SIGNED_URL), createResultExceptionHandler))
+        .setPutExceptionHandler(writeResultExceptionHandler)
+        .build();
   }
 
   private static ByteBuffer randomBuffer(int size) {
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/BucketTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/BucketTest.java
index e8a8e3f4..5d9e622f 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/BucketTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/BucketTest.java
@@ -136,6 +136,8 @@ public class BucketTest {
   private static final String BASE64_KEY = "JVzfVl8NLD9FjedFuStegjRfES5ll5zc59CIXw572OA=";
   private static final Key KEY =
       new SecretKeySpec(BaseEncoding.base64().decode(BASE64_KEY), "AES256");
+  private final RetryAlgorithmManager retryAlgorithmManager =
+      StorageOptions.getDefaultInstance().getRetryAlgorithmManager();
 
   private Storage storage;
   private Storage serviceMockReturnsOptions = createMock(Storage.class);
@@ -154,9 +156,11 @@ public class BucketTest {
     verify(storage);
   }
 
-  private void initializeExpectedBucket(int optionsCalls) {
-    expect(serviceMockReturnsOptions.getOptions()).andReturn(mockOptions).times(optionsCalls);
+  private void initializeExpectedBucket() {
+    expect(serviceMockReturnsOptions.getOptions()).andReturn(mockOptions).anyTimes();
     replay(serviceMockReturnsOptions);
+    expect(mockOptions.getRetryAlgorithmManager()).andReturn(retryAlgorithmManager).anyTimes();
+    replay(mockOptions);
     expectedBucket = new Bucket(serviceMockReturnsOptions, new BucketInfo.BuilderImpl(BUCKET_INFO));
     blobResults =
         ImmutableList.of(
@@ -177,7 +181,7 @@ public class BucketTest {
 
   @Test
   public void testExists_True() throws Exception {
-    initializeExpectedBucket(4);
+    initializeExpectedBucket();
     Storage.BucketGetOption[] expectedOptions = {Storage.BucketGetOption.fields()};
     expect(storage.getOptions()).andReturn(mockOptions);
     expect(storage.get(BUCKET_INFO.getName(), expectedOptions)).andReturn(expectedBucket);
@@ -188,7 +192,7 @@ public class BucketTest {
 
   @Test
   public void testExists_False() throws Exception {
-    initializeExpectedBucket(4);
+    initializeExpectedBucket();
     Storage.BucketGetOption[] expectedOptions = {Storage.BucketGetOption.fields()};
     expect(storage.getOptions()).andReturn(mockOptions);
     expect(storage.get(BUCKET_INFO.getName(), expectedOptions)).andReturn(null);
@@ -199,7 +203,7 @@ public class BucketTest {
 
   @Test
   public void testReload() throws Exception {
-    initializeExpectedBucket(5);
+    initializeExpectedBucket();
     BucketInfo updatedInfo = BUCKET_INFO.toBuilder().setNotFoundPage("p").build();
     Bucket expectedUpdatedBucket =
         new Bucket(serviceMockReturnsOptions, new BucketInfo.BuilderImpl(updatedInfo));
@@ -213,7 +217,7 @@ public class BucketTest {
 
   @Test
   public void testReloadNull() throws Exception {
-    initializeExpectedBucket(4);
+    initializeExpectedBucket();
     expect(storage.getOptions()).andReturn(mockOptions);
     expect(storage.get(BUCKET_INFO.getName())).andReturn(null);
     replay(storage);
@@ -223,7 +227,7 @@ public class BucketTest {
 
   @Test
   public void testReloadWithOptions() throws Exception {
-    initializeExpectedBucket(5);
+    initializeExpectedBucket();
     BucketInfo updatedInfo = BUCKET_INFO.toBuilder().setNotFoundPage("p").build();
     Bucket expectedUpdatedBucket =
         new Bucket(serviceMockReturnsOptions, new BucketInfo.BuilderImpl(updatedInfo));
@@ -238,7 +242,7 @@ public class BucketTest {
 
   @Test
   public void testUpdate() throws Exception {
-    initializeExpectedBucket(5);
+    initializeExpectedBucket();
     Bucket expectedUpdatedBucket = expectedBucket.toBuilder().setNotFoundPage("p").build();
     expect(storage.getOptions()).andReturn(mockOptions).times(2);
     expect(storage.update(expectedUpdatedBucket)).andReturn(expectedUpdatedBucket);
@@ -251,7 +255,7 @@ public class BucketTest {
 
   @Test
   public void testDelete() throws Exception {
-    initializeExpectedBucket(4);
+    initializeExpectedBucket();
     expect(storage.getOptions()).andReturn(mockOptions);
     expect(storage.delete(BUCKET_INFO.getName())).andReturn(true);
     replay(storage);
@@ -261,7 +265,7 @@ public class BucketTest {
 
   @Test
   public void testList() throws Exception {
-    initializeExpectedBucket(4);
+    initializeExpectedBucket();
     PageImpl<Blob> expectedBlobPage = new PageImpl<>(null, "c", blobResults);
     expect(storage.getOptions()).andReturn(mockOptions);
     expect(storage.list(BUCKET_INFO.getName())).andReturn(expectedBlobPage);
@@ -280,7 +284,7 @@ public class BucketTest {
 
   @Test
   public void testGet() throws Exception {
-    initializeExpectedBucket(5);
+    initializeExpectedBucket();
     Blob expectedBlob =
         new Blob(
             serviceMockReturnsOptions,
@@ -296,7 +300,7 @@ public class BucketTest {
 
   @Test
   public void testGetAllArray() throws Exception {
-    initializeExpectedBucket(4);
+    initializeExpectedBucket();
     expect(storage.getOptions()).andReturn(mockOptions);
     List<BlobId> blobIds =
         Lists.transform(
@@ -315,7 +319,7 @@ public class BucketTest {
 
   @Test
   public void testGetAllIterable() throws Exception {
-    initializeExpectedBucket(4);
+    initializeExpectedBucket();
     expect(storage.getOptions()).andReturn(mockOptions);
     List<BlobId> blobIds =
         Lists.transform(
@@ -334,7 +338,7 @@ public class BucketTest {
 
   @Test
   public void testCreate() throws Exception {
-    initializeExpectedBucket(5);
+    initializeExpectedBucket();
     BlobInfo info = BlobInfo.newBuilder("b", "n").setContentType(CONTENT_TYPE).build();
     Blob expectedBlob = new Blob(serviceMockReturnsOptions, new BlobInfo.BuilderImpl(info));
     byte[] content = {0xD, 0xE, 0xA, 0xD};
@@ -348,7 +352,7 @@ public class BucketTest {
 
   @Test
   public void testCreateNoContentType() throws Exception {
-    initializeExpectedBucket(5);
+    initializeExpectedBucket();
     BlobInfo info = BlobInfo.newBuilder("b", "n").build();
     Blob expectedBlob = new Blob(serviceMockReturnsOptions, new BlobInfo.BuilderImpl(info));
     byte[] content = {0xD, 0xE, 0xA, 0xD};
@@ -362,7 +366,7 @@ public class BucketTest {
 
   @Test
   public void testCreateWithOptions() throws Exception {
-    initializeExpectedBucket(5);
+    initializeExpectedBucket();
     BlobInfo info =
         BlobInfo.newBuilder(BlobId.of("b", "n", 42L))
             .setContentType(CONTENT_TYPE)
@@ -399,7 +403,7 @@ public class BucketTest {
 
   @Test
   public void testCreateWithEncryptionKey() throws Exception {
-    initializeExpectedBucket(5);
+    initializeExpectedBucket();
     BlobInfo info = BlobInfo.newBuilder(BlobId.of("b", "n")).setContentType(CONTENT_TYPE).build();
     Blob expectedBlob = new Blob(serviceMockReturnsOptions, new BlobInfo.BuilderImpl(info));
     byte[] content = {0xD, 0xE, 0xA, 0xD};
@@ -415,7 +419,7 @@ public class BucketTest {
 
   @Test
   public void testCreateWithKmsKeyName() throws Exception {
-    initializeExpectedBucket(5);
+    initializeExpectedBucket();
     BlobInfo info = BlobInfo.newBuilder(BlobId.of("b", "n")).setContentType(CONTENT_TYPE).build();
     Blob expectedBlob = new Blob(serviceMockReturnsOptions, new BlobInfo.BuilderImpl(info));
     byte[] content = {0xD, 0xE, 0xA, 0xD};
@@ -432,7 +436,7 @@ public class BucketTest {
 
   @Test
   public void testCreateNotExists() throws Exception {
-    initializeExpectedBucket(5);
+    initializeExpectedBucket();
     BlobInfo info =
         BlobInfo.newBuilder(BlobId.of("b", "n", 0L)).setContentType(CONTENT_TYPE).build();
     Blob expectedBlob = new Blob(serviceMockReturnsOptions, new BlobInfo.BuilderImpl(info));
@@ -448,7 +452,7 @@ public class BucketTest {
 
   @Test
   public void testCreateWithWrongGenerationOptions() throws Exception {
-    initializeExpectedBucket(4);
+    initializeExpectedBucket();
     expect(storage.getOptions()).andReturn(mockOptions);
     replay(storage);
     initializeBucket();
@@ -468,7 +472,7 @@ public class BucketTest {
 
   @Test
   public void testCreateWithWrongMetagenerationOptions() throws Exception {
-    initializeExpectedBucket(4);
+    initializeExpectedBucket();
     expect(storage.getOptions()).andReturn(mockOptions);
     replay(storage);
     initializeBucket();
@@ -489,7 +493,7 @@ public class BucketTest {
   @Test
   @SuppressWarnings({"unchecked", "deprecation"})
   public void testCreateFromStream() throws Exception {
-    initializeExpectedBucket(5);
+    initializeExpectedBucket();
     BlobInfo info = BlobInfo.newBuilder("b", "n").setContentType(CONTENT_TYPE).build();
     Blob expectedBlob = new Blob(serviceMockReturnsOptions, new BlobInfo.BuilderImpl(info));
     byte[] content = {0xD, 0xE, 0xA, 0xD};
@@ -505,7 +509,7 @@ public class BucketTest {
   @Test
   @SuppressWarnings({"unchecked", "deprecation"})
   public void testCreateFromStreamNoContentType() throws Exception {
-    initializeExpectedBucket(5);
+    initializeExpectedBucket();
     BlobInfo info = BlobInfo.newBuilder("b", "n").build();
     Blob expectedBlob = new Blob(serviceMockReturnsOptions, new BlobInfo.BuilderImpl(info));
     byte[] content = {0xD, 0xE, 0xA, 0xD};
@@ -521,7 +525,7 @@ public class BucketTest {
   @Test
   @SuppressWarnings({"unchecked", "deprecation"})
   public void testCreateFromStreamWithOptions() throws Exception {
-    initializeExpectedBucket(5);
+    initializeExpectedBucket();
     BlobInfo info =
         BlobInfo.newBuilder(BlobId.of("b", "n", 42L))
             .setContentType(CONTENT_TYPE)
@@ -566,7 +570,7 @@ public class BucketTest {
   @Test
   @SuppressWarnings({"unchecked", "deprecation"})
   public void testCreateFromStreamWithEncryptionKey() throws Exception {
-    initializeExpectedBucket(5);
+    initializeExpectedBucket();
     BlobInfo info = BlobInfo.newBuilder(BlobId.of("b", "n")).setContentType(CONTENT_TYPE).build();
     Blob expectedBlob = new Blob(serviceMockReturnsOptions, new BlobInfo.BuilderImpl(info));
     byte[] content = {0xD, 0xE, 0xA, 0xD};
@@ -584,7 +588,7 @@ public class BucketTest {
   @Test
   @SuppressWarnings({"unchecked", "deprecation"})
   public void testCreateFromStreamNotExists() throws Exception {
-    initializeExpectedBucket(5);
+    initializeExpectedBucket();
     BlobInfo info =
         BlobInfo.newBuilder(BlobId.of("b", "n", 0L)).setContentType(CONTENT_TYPE).build();
     Blob expectedBlob = new Blob(serviceMockReturnsOptions, new BlobInfo.BuilderImpl(info));
@@ -602,7 +606,7 @@ public class BucketTest {
 
   @Test
   public void testCreateFromStreamWithWrongGenerationOptions() throws Exception {
-    initializeExpectedBucket(4);
+    initializeExpectedBucket();
     expect(storage.getOptions()).andReturn(mockOptions);
     replay(storage);
     initializeBucket();
@@ -623,7 +627,7 @@ public class BucketTest {
 
   @Test
   public void testCreateFromStreamWithWrongMetagenerationOptions() throws Exception {
-    initializeExpectedBucket(4);
+    initializeExpectedBucket();
     expect(storage.getOptions()).andReturn(mockOptions);
     replay(storage);
     initializeBucket();
@@ -644,7 +648,7 @@ public class BucketTest {
 
   @Test
   public void testGetAcl() throws Exception {
-    initializeExpectedBucket(4);
+    initializeExpectedBucket();
     expect(storage.getOptions()).andReturn(mockOptions);
     expect(storage.getAcl(BUCKET_INFO.getName(), User.ofAllAuthenticatedUsers())).andReturn(ACL);
     replay(storage);
@@ -654,7 +658,7 @@ public class BucketTest {
 
   @Test
   public void testDeleteAcl() throws Exception {
-    initializeExpectedBucket(4);
+    initializeExpectedBucket();
     expect(storage.getOptions()).andReturn(mockOptions);
     expect(storage.deleteAcl(BUCKET_INFO.getName(), User.ofAllAuthenticatedUsers()))
         .andReturn(true);
@@ -665,7 +669,7 @@ public class BucketTest {
 
   @Test
   public void testCreateAcl() throws Exception {
-    initializeExpectedBucket(4);
+    initializeExpectedBucket();
     expect(storage.getOptions()).andReturn(mockOptions);
     Acl returnedAcl = ACL.toBuilder().setEtag("ETAG").setId("ID").build();
     expect(storage.createAcl(BUCKET_INFO.getName(), ACL)).andReturn(returnedAcl);
@@ -676,7 +680,7 @@ public class BucketTest {
 
   @Test
   public void testUpdateAcl() throws Exception {
-    initializeExpectedBucket(4);
+    initializeExpectedBucket();
     expect(storage.getOptions()).andReturn(mockOptions);
     Acl returnedAcl = ACL.toBuilder().setEtag("ETAG").setId("ID").build();
     expect(storage.updateAcl(BUCKET_INFO.getName(), ACL)).andReturn(returnedAcl);
@@ -687,7 +691,7 @@ public class BucketTest {
 
   @Test
   public void testListAcls() throws Exception {
-    initializeExpectedBucket(4);
+    initializeExpectedBucket();
     expect(storage.getOptions()).andReturn(mockOptions);
     expect(storage.listAcls(BUCKET_INFO.getName())).andReturn(ACLS);
     replay(storage);
@@ -697,7 +701,7 @@ public class BucketTest {
 
   @Test
   public void testGetDefaultAcl() throws Exception {
-    initializeExpectedBucket(4);
+    initializeExpectedBucket();
     expect(storage.getOptions()).andReturn(mockOptions);
     expect(storage.getDefaultAcl(BUCKET_INFO.getName(), User.ofAllAuthenticatedUsers()))
         .andReturn(ACL);
@@ -708,7 +712,7 @@ public class BucketTest {
 
   @Test
   public void testDeleteDefaultAcl() throws Exception {
-    initializeExpectedBucket(4);
+    initializeExpectedBucket();
     expect(storage.getOptions()).andReturn(mockOptions);
     expect(storage.deleteDefaultAcl(BUCKET_INFO.getName(), User.ofAllAuthenticatedUsers()))
         .andReturn(true);
@@ -719,7 +723,7 @@ public class BucketTest {
 
   @Test
   public void testCreateDefaultAcl() throws Exception {
-    initializeExpectedBucket(4);
+    initializeExpectedBucket();
     expect(storage.getOptions()).andReturn(mockOptions);
     Acl returnedAcl = ACL.toBuilder().setEtag("ETAG").setId("ID").build();
     expect(storage.createDefaultAcl(BUCKET_INFO.getName(), ACL)).andReturn(returnedAcl);
@@ -730,7 +734,7 @@ public class BucketTest {
 
   @Test
   public void testUpdateDefaultAcl() throws Exception {
-    initializeExpectedBucket(4);
+    initializeExpectedBucket();
     expect(storage.getOptions()).andReturn(mockOptions);
     Acl returnedAcl = ACL.toBuilder().setEtag("ETAG").setId("ID").build();
     expect(storage.updateDefaultAcl(BUCKET_INFO.getName(), ACL)).andReturn(returnedAcl);
@@ -741,7 +745,7 @@ public class BucketTest {
 
   @Test
   public void testListDefaultAcls() throws Exception {
-    initializeExpectedBucket(4);
+    initializeExpectedBucket();
     expect(storage.getOptions()).andReturn(mockOptions);
     expect(storage.listDefaultAcls(BUCKET_INFO.getName())).andReturn(ACLS);
     replay(storage);
@@ -751,7 +755,7 @@ public class BucketTest {
 
   @Test
   public void testLockRetention() throws Exception {
-    initializeExpectedBucket(5);
+    initializeExpectedBucket();
     Bucket expectedRetentionLockedBucket =
         expectedBucket
             .toBuilder()
@@ -790,7 +794,7 @@ public class BucketTest {
   @Test
   @SuppressWarnings({"unchecked", "deprecation"})
   public void testBuilder() {
-    initializeExpectedBucket(4);
+    initializeExpectedBucket();
     expect(storage.getOptions()).andReturn(mockOptions).times(4);
     replay(storage);
     Bucket.Builder builder =
@@ -854,7 +858,7 @@ public class BucketTest {
 
   @Test
   public void testDeleteLifecycleRules() {
-    initializeExpectedBucket(6);
+    initializeExpectedBucket();
     Bucket bucket =
         new Bucket(serviceMockReturnsOptions, new BucketInfo.BuilderImpl(FULL_BUCKET_INFO));
     assertThat(bucket.getLifecycleRules()).hasSize(1);
@@ -870,7 +874,7 @@ public class BucketTest {
 
   @Test
   public void testUpdateBucketLogging() {
-    initializeExpectedBucket(6);
+    initializeExpectedBucket();
     BucketInfo.Logging logging =
         BucketInfo.Logging.newBuilder()
             .setLogBucket("logs-bucket")
@@ -893,7 +897,7 @@ public class BucketTest {
 
   @Test
   public void testRemoveBucketCORS() {
-    initializeExpectedBucket(6);
+    initializeExpectedBucket();
     List<Cors.Origin> origins = ImmutableList.of(Cors.Origin.of("http://cloud.google.com"));
     List<HttpMethod> httpMethods = ImmutableList.of(HttpMethod.GET);
     List<String> responseHeaders = ImmutableList.of("Content-Type");
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/PackagePrivateMethodWorkarounds.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/PackagePrivateMethodWorkarounds.java
index 0706df1a..fb86d5d1 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/PackagePrivateMethodWorkarounds.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/PackagePrivateMethodWorkarounds.java
@@ -38,4 +38,8 @@ public final class PackagePrivateMethodWorkarounds {
     BlobInfo.BuilderImpl builder = (BlobInfo.BuilderImpl) BlobInfo.fromPb(b.toPb()).toBuilder();
     return new Blob(s, builder);
   }
+
+  public static StorageOptions.Builder useNewRetryAlgorithmManager(StorageOptions.Builder builder) {
+    return builder.setUseDefaultRetryAlgorithms();
+  }
 }
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/ResumableMediaTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/ResumableMediaTest.java
new file mode 100644
index 00000000..47b165dc
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/ResumableMediaTest.java
@@ -0,0 +1,46 @@
+/*
+ * Copyright 2021 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.storage;
+
+import static org.junit.Assert.assertNotNull;
+
+import com.google.cloud.BaseService;
+import com.google.cloud.ExceptionHandler;
+import java.net.URL;
+import org.junit.Assert;
+import org.junit.Test;
+
+public final class ResumableMediaTest {
+  private static final String SIGNED_URL =
+      "http://www.test.com/test-bucket/test1.txt?GoogleAccessId=testClient-test@test.com&Expires=1553839761&Signature=MJUBXAZ7";
+
+  private final ExceptionHandler createResultExceptionHandler = BaseService.EXCEPTION_HANDLER;
+
+  @Test
+  public void startUploadForSignedUrl_expectStorageException_whenUrlInvalid() throws Exception {
+    try {
+      ResumableMedia.startUploadForSignedUrl(
+              StorageOptions.newBuilder().build(),
+              new URL(SIGNED_URL),
+              createResultExceptionHandler)
+          .get();
+      Assert.fail();
+    } catch (StorageException ex) {
+      assertNotNull(ex.getMessage());
+    }
+  }
+}
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/SerializationTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/SerializationTest.java
index ae49f13f..6d4876d4 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/SerializationTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/SerializationTest.java
@@ -17,6 +17,7 @@
 package com.google.cloud.storage;
 
 import com.google.cloud.BaseSerializationTest;
+import com.google.cloud.ExceptionHandler;
 import com.google.cloud.NoCredentials;
 import com.google.cloud.PageImpl;
 import com.google.cloud.ReadChannel;
@@ -99,12 +100,17 @@ public class SerializationTest extends BaseSerializationTest {
   @Override
   protected Restorable<?>[] restorableObjects() {
     StorageOptions options = StorageOptions.newBuilder().setProjectId("p2").build();
+    ExceptionHandler exceptionHandler =
+        options.getRetryAlgorithmManager().getForResumableUploadSessionWrite(EMPTY_RPC_OPTIONS);
     ReadChannel reader = new BlobReadChannel(options, BlobId.of("b", "n"), EMPTY_RPC_OPTIONS);
     // avoid closing when you don't want partial writes to GCS upon failure
     @SuppressWarnings("resource")
     BlobWriteChannel writer =
         new BlobWriteChannel(
-            options, BlobInfo.newBuilder(BlobId.of("b", "n")).build(), "upload-id");
+            options,
+            BlobInfo.newBuilder(BlobId.of("b", "n")).build(),
+            "upload-id",
+            exceptionHandler);
     return new Restorable<?>[] {reader, writer};
   }
 }
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageBatchTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageBatchTest.java
index 3943e140..978cffe0 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageBatchTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageBatchTest.java
@@ -54,6 +54,8 @@ public class StorageBatchTest {
     BlobTargetOption.generationMatch(), BlobTargetOption.metagenerationMatch()
   };
   private static final GoogleJsonError GOOGLE_JSON_ERROR = new GoogleJsonError();
+  private final RetryAlgorithmManager retryAlgorithmManager =
+      StorageOptions.getDefaultInstance().getRetryAlgorithmManager();
 
   private StorageOptions optionsMock;
   private StorageRpc storageRpcMock;
@@ -67,6 +69,9 @@ public class StorageBatchTest {
     storageRpcMock = EasyMock.createMock(StorageRpc.class);
     batchMock = EasyMock.createMock(RpcBatch.class);
     EasyMock.expect(optionsMock.getStorageRpcV1()).andReturn(storageRpcMock);
+    EasyMock.expect(optionsMock.getRetryAlgorithmManager())
+        .andReturn(retryAlgorithmManager)
+        .anyTimes();
     EasyMock.expect(storageRpcMock.createBatch()).andReturn(batchMock);
     EasyMock.replay(optionsMock, storageRpcMock, batchMock, storage);
     storageBatch = new StorageBatch(optionsMock);
@@ -165,8 +170,11 @@ public class StorageBatchTest {
   @Test
   public void testUpdateWithOptions() {
     EasyMock.reset(storage, batchMock, optionsMock);
-    EasyMock.expect(storage.getOptions()).andReturn(optionsMock).times(2);
-    EasyMock.expect(optionsMock.getService()).andReturn(storage);
+    EasyMock.expect(storage.getOptions()).andReturn(optionsMock).anyTimes();
+    EasyMock.expect(optionsMock.getService()).andReturn(storage).anyTimes();
+    EasyMock.expect(optionsMock.getRetryAlgorithmManager())
+        .andReturn(retryAlgorithmManager)
+        .anyTimes();
     Capture<RpcBatch.Callback<StorageObject>> callback = Capture.newInstance();
     Capture<Map<StorageRpc.Option, Object>> capturedOptions = Capture.newInstance();
     batchMock.addPatch(
@@ -216,8 +224,11 @@ public class StorageBatchTest {
   @Test
   public void testGetWithOptions() {
     EasyMock.reset(storage, batchMock, optionsMock);
-    EasyMock.expect(storage.getOptions()).andReturn(optionsMock).times(2);
-    EasyMock.expect(optionsMock.getService()).andReturn(storage);
+    EasyMock.expect(storage.getOptions()).andReturn(optionsMock).anyTimes();
+    EasyMock.expect(optionsMock.getService()).andReturn(storage).anyTimes();
+    EasyMock.expect(optionsMock.getRetryAlgorithmManager())
+        .andReturn(retryAlgorithmManager)
+        .anyTimes();
     Capture<RpcBatch.Callback<StorageObject>> callback = Capture.newInstance();
     Capture<Map<StorageRpc.Option, Object>> capturedOptions = Capture.newInstance();
     batchMock.addGet(
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageImplTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageImplTest.java
index 590fe37a..5619e5f7 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageImplTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageImplTest.java
@@ -38,7 +38,6 @@ import com.google.auth.oauth2.ServiceAccountCredentials;
 import com.google.cloud.Identity;
 import com.google.cloud.Policy;
 import com.google.cloud.ServiceOptions;
-import com.google.cloud.WriteChannel;
 import com.google.cloud.storage.Acl.Project;
 import com.google.cloud.storage.Acl.Project.ProjectRole;
 import com.google.cloud.storage.Acl.Role;
@@ -55,7 +54,6 @@ import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.io.BaseEncoding;
 import java.io.UnsupportedEncodingException;
-import java.net.MalformedURLException;
 import java.net.URL;
 import java.net.URLDecoder;
 import java.security.InvalidKeyException;
@@ -2148,16 +2146,6 @@ public class StorageImplTest {
     }
   }
 
-  @Test
-  public void testWriterWithSignedURL() throws MalformedURLException {
-    EasyMock.expect(storageRpcMock.open(SIGNED_URL)).andReturn("upload-id");
-    EasyMock.replay(storageRpcMock);
-    initializeService();
-    WriteChannel writer = new BlobWriteChannel(options, new URL(SIGNED_URL));
-    assertNotNull(writer);
-    assertTrue(writer.isOpen());
-  }
-
   @Test
   public void testV4PostPolicy() {
     EasyMock.replay(storageRpcMock);
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/conformance/retry/RetryTestFixture.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/conformance/retry/RetryTestFixture.java
index cb0415c1..79a19a12 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/conformance/retry/RetryTestFixture.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/conformance/retry/RetryTestFixture.java
@@ -23,6 +23,7 @@ import com.google.api.gax.rpc.FixedHeaderProvider;
 import com.google.cloud.NoCredentials;
 import com.google.cloud.conformance.storage.v1.InstructionList;
 import com.google.cloud.conformance.storage.v1.Method;
+import com.google.cloud.storage.PackagePrivateMethodWorkarounds;
 import com.google.cloud.storage.Storage;
 import com.google.cloud.storage.StorageOptions;
 import com.google.cloud.storage.conformance.retry.TestBench.RetryTestResource;
@@ -138,6 +139,7 @@ final class RetryTestFixture implements TestRule {
             .setHost(testBench.getBaseUri())
             .setCredentials(NoCredentials.getInstance())
             .setProjectId("conformance-tests");
+    builder = PackagePrivateMethodWorkarounds.useNewRetryAlgorithmManager(builder);
     if (forTest) {
       builder.setHeaderProvider(
           new FixedHeaderProvider() {