Repository: googleapis/java-storage
Pull request #1622 (2022-09-15 17:40:24)

feat: implement automatic retry for resumable uploads
Plumb through the necessary items to allow retrying to take place in WriteStrategy.FlushOnEveryFlusher#flush.
Update corresponding builders along with this to pass through the necessary items.

### Refactorings
* Move {Unbuffered,Buffered}{Direct,Resumable}UploadBuilder classes to be nested in their logical "parent builders" to allow closing over fields set at the higher level
* Move GrpcTransformPageDecoratorTest.java#retryingDeps -> TestUtils#defaultRetryingDeps


BenWhitehead: Thanks for the review!

Patch:
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedWritableByteChannel.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedWritableByteChannel.java
index 442335b6..c4e15733 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedWritableByteChannel.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUnbufferedWritableByteChannel.java
@@ -23,6 +23,7 @@ import com.google.cloud.storage.UnbufferedWritableByteChannelSession.UnbufferedW
 import com.google.cloud.storage.WriteCtx.WriteObjectRequestBuilderFactory;
 import com.google.cloud.storage.WriteFlushStrategy.Flusher;
 import com.google.cloud.storage.WriteFlushStrategy.FlusherFactory;
+import com.google.common.annotations.VisibleForTesting;
 import com.google.protobuf.ByteString;
 import com.google.storage.v2.ChecksummedData;
 import com.google.storage.v2.ObjectChecksums;
@@ -58,9 +59,7 @@ final class GapicUnbufferedWritableByteChannel<
     this.writeCtx = new WriteCtx<>(requestFactory);
     this.flusher =
         flusherFactory.newFlusher(
-            requestFactory.bucketName(),
-            writeCtx.getConfirmedBytes()::addAndGet,
-            resultFuture::set);
+            requestFactory.bucketName(), writeCtx.getConfirmedBytes()::set, resultFuture::set);
   }
 
   @Override
@@ -157,4 +156,9 @@ final class GapicUnbufferedWritableByteChannel<
     }
     open = false;
   }
+
+  @VisibleForTesting
+  WriteCtx<RequestFactoryT> getWriteCtx() {
+    return writeCtx;
+  }
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicWritableByteChannelSessionBuilder.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicWritableByteChannelSessionBuilder.java
index f34c1c67..a3b5b5ab 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicWritableByteChannelSessionBuilder.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicWritableByteChannelSessionBuilder.java
@@ -21,9 +21,11 @@ import static java.util.Objects.requireNonNull;
 import com.google.api.core.ApiFuture;
 import com.google.api.core.ApiFutures;
 import com.google.api.core.SettableApiFuture;
+import com.google.api.gax.retrying.ResultRetryAlgorithm;
 import com.google.api.gax.rpc.ClientStreamingCallable;
 import com.google.cloud.storage.ChannelSession.BufferedWriteSession;
 import com.google.cloud.storage.ChannelSession.UnbufferedWriteSession;
+import com.google.cloud.storage.Retrying.RetryingDependencies;
 import com.google.cloud.storage.UnbufferedWritableByteChannelSession.UnbufferedWritableByteChannel;
 import com.google.cloud.storage.WriteCtx.WriteObjectRequestBuilderFactory;
 import com.google.cloud.storage.WriteFlushStrategy.FlusherFactory;
@@ -165,10 +167,70 @@ final class GapicWritableByteChannelSessionBuilder {
     private BufferedDirectUploadBuilder buffered(BufferHandle bufferHandle) {
       return new BufferedDirectUploadBuilder(bufferHandle);
     }
+
+    final class UnbufferedDirectUploadBuilder {
+
+      private WriteObjectRequest req;
+
+      /** Specify the {@link WriteObjectRequest} which will be used to start the Write stream. */
+      UnbufferedDirectUploadBuilder setRequest(WriteObjectRequest req) {
+        this.req = requireNonNull(req, "req must be non null");
+        return this;
+      }
+
+      UnbufferedWritableByteChannelSession<WriteObjectResponse> build() {
+        return new UnbufferedWriteSession<>(
+            ApiFutures.immediateFuture(requireNonNull(req, "req must be non null")),
+            bindFunction(
+                    WriteFlushStrategy.fsyncOnClose(write),
+                    WriteObjectRequestBuilderFactory::simple)
+                .andThen(StorageByteChannels.writable()::createSynchronized));
+      }
+    }
+
+    final class BufferedDirectUploadBuilder {
+
+      private final BufferHandle bufferHandle;
+      private WriteObjectRequest req;
+
+      BufferedDirectUploadBuilder(BufferHandle bufferHandle) {
+        this.bufferHandle = bufferHandle;
+      }
+
+      /** Specify the {@link WriteObjectRequest} which will be used to start the Write stream. */
+      BufferedDirectUploadBuilder setRequest(WriteObjectRequest req) {
+        this.req = requireNonNull(req, "req must be non null");
+        return this;
+      }
+
+      BufferedWritableByteChannelSession<WriteObjectResponse> build() {
+        return new BufferedWriteSession<>(
+            ApiFutures.immediateFuture(requireNonNull(req, "req must be non null")),
+            bindFunction(
+                    WriteFlushStrategy.fsyncOnClose(write),
+                    WriteObjectRequestBuilderFactory::simple)
+                .andThen(c -> new DefaultBufferedWritableByteChannel(bufferHandle, c))
+                .andThen(StorageByteChannels.writable()::createSynchronized));
+      }
+    }
   }
 
   final class ResumableUploadBuilder {
 
+    private RetryingDependencies deps;
+    private ResultRetryAlgorithm<?> alg;
+
+    ResumableUploadBuilder() {
+      this.deps = RetryingDependencies.attemptOnce();
+      this.alg = Retrying.neverRetry();
+    }
+
+    ResumableUploadBuilder withRetryConfig(RetryingDependencies deps, ResultRetryAlgorithm<?> alg) {
+      this.deps = requireNonNull(deps, "deps must be non null");
+      this.alg = requireNonNull(alg, "alg must be non null");
+      return this;
+    }
+
     /**
      * Do not apply any intermediate buffering. Any call to {@link
      * java.nio.channels.WritableByteChannel#write(ByteBuffer)} will be segmented as is and sent to
@@ -201,98 +263,56 @@ final class GapicWritableByteChannelSessionBuilder {
     private BufferedResumableUploadBuilder buffered(BufferHandle bufferHandle) {
       return new BufferedResumableUploadBuilder(bufferHandle);
     }
-  }
-
-  final class UnbufferedDirectUploadBuilder {
-
-    private WriteObjectRequest req;
-
-    /** Specify the {@link WriteObjectRequest} which will be used to start the Write stream. */
-    UnbufferedDirectUploadBuilder setRequest(WriteObjectRequest req) {
-      this.req = requireNonNull(req, "req must be non null");
-      return this;
-    }
-
-    UnbufferedWritableByteChannelSession<WriteObjectResponse> build() {
-      return new UnbufferedWriteSession<>(
-          ApiFutures.immediateFuture(requireNonNull(req, "req must be non null")),
-          bindFunction(
-                  WriteFlushStrategy.fsyncOnClose(write), WriteObjectRequestBuilderFactory::simple)
-              .andThen(StorageByteChannels.writable()::createSynchronized));
-    }
-  }
-
-  final class BufferedDirectUploadBuilder {
-
-    private final BufferHandle bufferHandle;
-    private WriteObjectRequest req;
-
-    BufferedDirectUploadBuilder(BufferHandle bufferHandle) {
-      this.bufferHandle = bufferHandle;
-    }
-
-    /** Specify the {@link WriteObjectRequest} which will be used to start the Write stream. */
-    BufferedDirectUploadBuilder setRequest(WriteObjectRequest req) {
-      this.req = requireNonNull(req, "req must be non null");
-      return this;
-    }
-
-    BufferedWritableByteChannelSession<WriteObjectResponse> build() {
-      return new BufferedWriteSession<>(
-          ApiFutures.immediateFuture(requireNonNull(req, "req must be non null")),
-          bindFunction(
-                  WriteFlushStrategy.fsyncOnClose(write), WriteObjectRequestBuilderFactory::simple)
-              .andThen(c -> new DefaultBufferedWritableByteChannel(bufferHandle, c))
-              .andThen(StorageByteChannels.writable()::createSynchronized));
-    }
-  }
 
-  final class UnbufferedResumableUploadBuilder {
-
-    private ApiFuture<ResumableWrite> start;
-
-    /**
-     * Set the Future which will contain the ResumableWrite information necessary to open the Write
-     * stream.
-     */
-    UnbufferedResumableUploadBuilder setStartAsync(ApiFuture<ResumableWrite> start) {
-      this.start = requireNonNull(start, "start must be non null");
-      return this;
-    }
-
-    UnbufferedWritableByteChannelSession<WriteObjectResponse> build() {
-      return new UnbufferedWriteSession<>(
-          requireNonNull(start, "start must be non null"),
-          bindFunction(WriteFlushStrategy.fsyncEveryFlush(write), ResumableWrite::identity)
-              .andThen(StorageByteChannels.writable()::createSynchronized));
-    }
-  }
-
-  final class BufferedResumableUploadBuilder {
-
-    private final BufferHandle bufferHandle;
-
-    private ApiFuture<ResumableWrite> start;
-
-    BufferedResumableUploadBuilder(BufferHandle bufferHandle) {
-      this.bufferHandle = bufferHandle;
-    }
-
-    /**
-     * Set the Future which will contain the ResumableWrite information necessary to open the Write
-     * stream.
-     */
-    BufferedResumableUploadBuilder setStartAsync(ApiFuture<ResumableWrite> start) {
-      this.start = requireNonNull(start, "start must be non null");
-      return this;
+    final class UnbufferedResumableUploadBuilder {
+
+      private ApiFuture<ResumableWrite> start;
+
+      /**
+       * Set the Future which will contain the ResumableWrite information necessary to open the
+       * Write stream.
+       */
+      UnbufferedResumableUploadBuilder setStartAsync(ApiFuture<ResumableWrite> start) {
+        this.start = requireNonNull(start, "start must be non null");
+        return this;
+      }
+
+      UnbufferedWritableByteChannelSession<WriteObjectResponse> build() {
+        return new UnbufferedWriteSession<>(
+            requireNonNull(start, "start must be non null"),
+            bindFunction(
+                    WriteFlushStrategy.fsyncEveryFlush(write, deps, alg), ResumableWrite::identity)
+                .andThen(StorageByteChannels.writable()::createSynchronized));
+      }
     }
 
-    BufferedWritableByteChannelSession<WriteObjectResponse> build() {
-      return new BufferedWriteSession<>(
-          requireNonNull(start, "start must be non null"),
-          bindFunction(WriteFlushStrategy.fsyncEveryFlush(write), ResumableWrite::identity)
-              .andThen(c -> new DefaultBufferedWritableByteChannel(bufferHandle, c))
-              .andThen(StorageByteChannels.writable()::createSynchronized));
+    final class BufferedResumableUploadBuilder {
+
+      private final BufferHandle bufferHandle;
+
+      private ApiFuture<ResumableWrite> start;
+
+      BufferedResumableUploadBuilder(BufferHandle bufferHandle) {
+        this.bufferHandle = bufferHandle;
+      }
+
+      /**
+       * Set the Future which will contain the ResumableWrite information necessary to open the
+       * Write stream.
+       */
+      BufferedResumableUploadBuilder setStartAsync(ApiFuture<ResumableWrite> start) {
+        this.start = requireNonNull(start, "start must be non null");
+        return this;
+      }
+
+      BufferedWritableByteChannelSession<WriteObjectResponse> build() {
+        return new BufferedWriteSession<>(
+            requireNonNull(start, "start must be non null"),
+            bindFunction(
+                    WriteFlushStrategy.fsyncEveryFlush(write, deps, alg), ResumableWrite::identity)
+                .andThen(c -> new DefaultBufferedWritableByteChannel(bufferHandle, c))
+                .andThen(StorageByteChannels.writable()::createSynchronized));
+      }
     }
   }
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcBlobWriteChannel.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcBlobWriteChannel.java
index 4ac3b3ac..de394466 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcBlobWriteChannel.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcBlobWriteChannel.java
@@ -20,10 +20,12 @@ import static com.google.cloud.storage.ByteSizeConstants._15MiB;
 import static com.google.cloud.storage.ByteSizeConstants._256KiB;
 
 import com.google.api.core.ApiFuture;
+import com.google.api.gax.retrying.ResultRetryAlgorithm;
 import com.google.api.gax.rpc.ClientStreamingCallable;
 import com.google.cloud.RestorableState;
 import com.google.cloud.WriteChannel;
 import com.google.cloud.storage.BufferedWritableByteChannelSession.BufferedWritableByteChannel;
+import com.google.cloud.storage.Retrying.RetryingDependencies;
 import com.google.common.base.Preconditions;
 import com.google.common.base.Suppliers;
 import com.google.storage.v2.WriteObjectRequest;
@@ -40,6 +42,8 @@ final class GrpcBlobWriteChannel implements WriteChannel {
 
   GrpcBlobWriteChannel(
       ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write,
+      RetryingDependencies deps,
+      ResultRetryAlgorithm<?> alg,
       Supplier<ApiFuture<ResumableWrite>> start) {
     lazyWriteChannel =
         new LazyWriteChannel(
@@ -51,6 +55,7 @@ final class GrpcBlobWriteChannel implements WriteChannel {
                         .setHasher(Hasher.noop())
                         .setByteStringStrategy(ByteStringStrategy.copy())
                         .resumable()
+                        .withRetryConfig(deps, alg)
                         .buffered(Buffers.allocateAligned(chunkSize, _256KiB))
                         .setStartAsync(start.get())
                         .build()));
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java
index cced65c8..1d35c739 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java
@@ -300,6 +300,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
       session =
           channelSessionBuilder
               .resumable()
+              .withRetryConfig(getOptions(), retryAlgorithmManager.idempotent())
               .buffered(Buffers.allocateAligned(bufferSize, _256KiB))
               .setStartAsync(start)
               .build();
@@ -346,6 +347,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
             .setHasher(Hasher.enabled())
             .setByteStringStrategy(ByteStringStrategy.noCopy())
             .resumable()
+            .withRetryConfig(getOptions(), retryAlgorithmManager.idempotent())
             .buffered(Buffers.allocateAligned(bufferSize, _256KiB))
             .setStartAsync(start)
             .build();
@@ -715,6 +717,8 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
     WriteObjectRequest req = getWriteObjectRequest(blobInfo, opts);
     return new GrpcBlobWriteChannel(
         storageClient.writeObjectCallable(),
+        getOptions(),
+        retryAlgorithmManager.idempotent(),
         () ->
             ResumableMedia.gapic()
                 .write()
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 e529ee9f..0740d120 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
@@ -19,6 +19,8 @@ package com.google.cloud.storage;
 import static com.google.cloud.RetryHelper.runWithRetries;
 
 import com.google.api.core.ApiClock;
+import com.google.api.core.NanoClock;
+import com.google.api.gax.retrying.BasicResultRetryAlgorithm;
 import com.google.api.gax.retrying.ResultRetryAlgorithm;
 import com.google.api.gax.retrying.RetrySettings;
 import com.google.cloud.RetryHelper.RetryHelperException;
@@ -95,6 +97,15 @@ final class Retrying {
     }
   }
 
+  static ResultRetryAlgorithm<?> neverRetry() {
+    return new BasicResultRetryAlgorithm<Object>() {
+      @Override
+      public boolean shouldRetry(Throwable previousThrowable, Object previousResponse) {
+        return false;
+      }
+    };
+  }
+
   /**
    * Rather than requiring a full set of {@link StorageOptions} to be passed specify what we
    * actually need and have StorageOptions implement this interface.
@@ -104,5 +115,19 @@ final class Retrying {
     RetrySettings getRetrySettings();
 
     ApiClock getClock();
+
+    static RetryingDependencies attemptOnce() {
+      return new RetryingDependencies() {
+        @Override
+        public RetrySettings getRetrySettings() {
+          return RetrySettings.newBuilder().setMaxAttempts(1).build();
+        }
+
+        @Override
+        public ApiClock getClock() {
+          return NanoClock.getDefaultClock();
+        }
+      };
+    }
   }
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/WriteFlushStrategy.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/WriteFlushStrategy.java
index 6da9c5e3..0007ec68 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/WriteFlushStrategy.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/WriteFlushStrategy.java
@@ -18,8 +18,11 @@ package com.google.cloud.storage;
 
 import com.google.api.core.SettableApiFuture;
 import com.google.api.gax.grpc.GrpcCallContext;
+import com.google.api.gax.retrying.ResultRetryAlgorithm;
 import com.google.api.gax.rpc.ApiStreamObserver;
 import com.google.api.gax.rpc.ClientStreamingCallable;
+import com.google.cloud.storage.Conversions.Decoder;
+import com.google.cloud.storage.Retrying.RetryingDependencies;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.storage.v2.WriteObjectRequest;
@@ -52,11 +55,14 @@ final class WriteFlushStrategy {
    * along with {@link Flusher#close(WriteObjectRequest)}.
    */
   static FlusherFactory fsyncEveryFlush(
-      ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write) {
+      ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write,
+      RetryingDependencies deps,
+      ResultRetryAlgorithm<?> alg) {
     return (String bucketName,
         LongConsumer committedTotalBytesCallback,
         Consumer<WriteObjectResponse> onSuccessCallback) ->
-        new FsyncEveryFlusher(write, bucketName, committedTotalBytesCallback, onSuccessCallback);
+        new FsyncEveryFlusher(
+            write, deps, alg, bucketName, committedTotalBytesCallback, onSuccessCallback);
   }
 
   /**
@@ -105,39 +111,51 @@ final class WriteFlushStrategy {
   private static final class FsyncEveryFlusher implements Flusher {
 
     private final ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write;
+    private final RetryingDependencies deps;
+    private final ResultRetryAlgorithm<?> alg;
     private final String bucketName;
     private final LongConsumer sizeCallback;
     private final Consumer<WriteObjectResponse> completeCallback;
 
     private FsyncEveryFlusher(
         ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write,
+        RetryingDependencies deps,
+        ResultRetryAlgorithm<?> alg,
         String bucketName,
         LongConsumer sizeCallback,
         Consumer<WriteObjectResponse> completeCallback) {
       this.write = write;
+      this.deps = deps;
+      this.alg = alg;
       this.bucketName = bucketName;
       this.sizeCallback = sizeCallback;
       this.completeCallback = completeCallback;
     }
 
     public void flush(@NonNull List<WriteObjectRequest> segments) {
-
-      Observer observer = new Observer(sizeCallback, completeCallback);
-      GrpcCallContext internalContext = contextWithBucketName(bucketName);
-      ApiStreamObserver<WriteObjectRequest> write =
-          this.write.withDefaultCallContext(internalContext).clientStreamingCall(observer);
-
-      boolean first = true;
-      for (WriteObjectRequest message : segments) {
-        if (!first) {
-          message = message.toBuilder().clearUploadId().clearWriteObjectSpec().build();
-        }
-
-        write.onNext(message);
-        first = false;
-      }
-      write.onCompleted();
-      observer.await();
+      Retrying.run(
+          deps,
+          alg,
+          () -> {
+            Observer observer = new Observer(sizeCallback, completeCallback);
+            GrpcCallContext internalContext = contextWithBucketName(bucketName);
+            ApiStreamObserver<WriteObjectRequest> write =
+                this.write.withDefaultCallContext(internalContext).clientStreamingCall(observer);
+
+            boolean first = true;
+            for (WriteObjectRequest message : segments) {
+              if (!first) {
+                message = message.toBuilder().clearUploadId().clearWriteObjectSpec().build();
+              }
+
+              write.onNext(message);
+              first = false;
+            }
+            write.onCompleted();
+            observer.await();
+            return null;
+          },
+          Decoder.identity());
     }
 
     public void close(@Nullable WriteObjectRequest req) {
@@ -243,7 +261,6 @@ final class WriteFlushStrategy {
      */
     @Override
     public void onError(Throwable t) {
-      // TODO: is retryable?
       invocationHandle.setException(t);
     }
 


Test patch:
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/GapicUnbufferedWritableByteChannelTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/GapicUnbufferedWritableByteChannelTest.java
index 70ad6a34..c9f122fb 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/GapicUnbufferedWritableByteChannelTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/GapicUnbufferedWritableByteChannelTest.java
@@ -16,9 +16,13 @@
 
 package com.google.cloud.storage;
 
+import static com.google.cloud.storage.TestUtils.apiException;
 import static com.google.common.truth.Truth.assertThat;
 
 import com.google.api.core.SettableApiFuture;
+import com.google.api.gax.retrying.BasicResultRetryAlgorithm;
+import com.google.api.gax.rpc.DataLossException;
+import com.google.cloud.storage.Retrying.RetryingDependencies;
 import com.google.cloud.storage.WriteCtx.WriteObjectRequestBuilderFactory;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
@@ -38,11 +42,70 @@ import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.util.List;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.BiConsumer;
 import org.junit.Test;
 
 public final class GapicUnbufferedWritableByteChannelTest {
 
-  ChunkSegmenter segmenter = new ChunkSegmenter(Hasher.noop(), ByteStringStrategy.copy(), 10, 5);
+  private static final ChunkSegmenter segmenter =
+      new ChunkSegmenter(Hasher.noop(), ByteStringStrategy.copy(), 10, 5);
+
+  private static final String uploadId = "upload-id";
+
+  private static final Object obj = Object.newBuilder().setBucket("buck").setName("obj").build();
+  private static final WriteObjectSpec spec = WriteObjectSpec.newBuilder().setResource(obj).build();
+
+  private static final StartResumableWriteRequest startReq =
+      StartResumableWriteRequest.newBuilder().setWriteObjectSpec(spec).build();
+  private static final StartResumableWriteResponse startResp =
+      StartResumableWriteResponse.newBuilder().setUploadId(uploadId).build();
+
+  private static final byte[] bytes = DataGenerator.base64Characters().genBytes(40);
+  private static final WriteObjectRequest req1 =
+      WriteObjectRequest.newBuilder()
+          .setUploadId(uploadId)
+          .setChecksummedData(TestUtils.getChecksummedData(ByteString.copyFrom(bytes, 0, 10)))
+          .build();
+  private static final WriteObjectRequest req2 =
+      WriteObjectRequest.newBuilder()
+          .setUploadId(uploadId)
+          .setWriteOffset(10)
+          .setChecksummedData(TestUtils.getChecksummedData(ByteString.copyFrom(bytes, 10, 10)))
+          .build();
+  private static final WriteObjectRequest req3 =
+      WriteObjectRequest.newBuilder()
+          .setUploadId(uploadId)
+          .setWriteOffset(20)
+          .setChecksummedData(TestUtils.getChecksummedData(ByteString.copyFrom(bytes, 20, 10)))
+          .build();
+  private static final WriteObjectRequest req4 =
+      WriteObjectRequest.newBuilder()
+          .setUploadId(uploadId)
+          .setWriteOffset(30)
+          .setChecksummedData(TestUtils.getChecksummedData(ByteString.copyFrom(bytes, 30, 10)))
+          .build();
+  private static final WriteObjectRequest req5 =
+      WriteObjectRequest.newBuilder()
+          .setUploadId(uploadId)
+          .setWriteOffset(40)
+          .setFinishWrite(true)
+          .build();
+
+  private static final WriteObjectResponse resp1 =
+      WriteObjectResponse.newBuilder().setPersistedSize(10).build();
+  private static final WriteObjectResponse resp2 =
+      WriteObjectResponse.newBuilder().setPersistedSize(20).build();
+  private static final WriteObjectResponse resp3 =
+      WriteObjectResponse.newBuilder().setPersistedSize(30).build();
+  private static final WriteObjectResponse resp4 =
+      WriteObjectResponse.newBuilder().setPersistedSize(40).build();
+  private static final WriteObjectResponse resp5 =
+      WriteObjectResponse.newBuilder().setResource(obj.toBuilder().setSize(40)).build();
+
+  private static final WriteObjectRequestBuilderFactory reqFactory =
+      new ResumableWrite(startReq, startResp);
 
   @Test
   public void directUpload() throws IOException, InterruptedException, ExecutionException {
@@ -99,56 +162,6 @@ public final class GapicUnbufferedWritableByteChannelTest {
 
   @Test
   public void resumableUpload() throws IOException, InterruptedException, ExecutionException {
-    String uploadId = "upload-id";
-
-    Object obj = Object.newBuilder().setBucket("buck").setName("obj").build();
-    WriteObjectSpec spec = WriteObjectSpec.newBuilder().setResource(obj).build();
-
-    StartResumableWriteRequest startReq =
-        StartResumableWriteRequest.newBuilder().setWriteObjectSpec(spec).build();
-    StartResumableWriteResponse startResp =
-        StartResumableWriteResponse.newBuilder().setUploadId(uploadId).build();
-
-    byte[] bytes = DataGenerator.base64Characters().genBytes(40);
-    WriteObjectRequest req1 =
-        WriteObjectRequest.newBuilder()
-            .setUploadId(uploadId)
-            .setChecksummedData(TestUtils.getChecksummedData(ByteString.copyFrom(bytes, 0, 10)))
-            .build();
-    WriteObjectRequest req2 =
-        WriteObjectRequest.newBuilder()
-            .setUploadId(uploadId)
-            .setWriteOffset(10)
-            .setChecksummedData(TestUtils.getChecksummedData(ByteString.copyFrom(bytes, 10, 10)))
-            .build();
-    WriteObjectRequest req3 =
-        WriteObjectRequest.newBuilder()
-            .setUploadId(uploadId)
-            .setWriteOffset(20)
-            .setChecksummedData(TestUtils.getChecksummedData(ByteString.copyFrom(bytes, 20, 10)))
-            .build();
-    WriteObjectRequest req4 =
-        WriteObjectRequest.newBuilder()
-            .setUploadId(uploadId)
-            .setWriteOffset(30)
-            .setChecksummedData(TestUtils.getChecksummedData(ByteString.copyFrom(bytes, 30, 10)))
-            .build();
-    WriteObjectRequest req5 =
-        WriteObjectRequest.newBuilder()
-            .setUploadId(uploadId)
-            .setWriteOffset(40)
-            .setFinishWrite(true)
-            .build();
-
-    WriteObjectResponse resp1 = WriteObjectResponse.newBuilder().setPersistedSize(10).build();
-    WriteObjectResponse resp2 = WriteObjectResponse.newBuilder().setPersistedSize(20).build();
-    WriteObjectResponse resp3 = WriteObjectResponse.newBuilder().setPersistedSize(30).build();
-    WriteObjectResponse resp4 = WriteObjectResponse.newBuilder().setPersistedSize(40).build();
-    WriteObjectResponse resp5 =
-        WriteObjectResponse.newBuilder().setResource(obj.toBuilder().setSize(40)).build();
-
-    WriteObjectRequestBuilderFactory reqFactory = new ResumableWrite(startReq, startResp);
-
     ImmutableMap<List<WriteObjectRequest>, WriteObjectResponse> writes =
         ImmutableMap.<List<WriteObjectRequest>, WriteObjectResponse>builder()
             .put(ImmutableList.of(req1), resp1)
@@ -166,7 +179,10 @@ public final class GapicUnbufferedWritableByteChannelTest {
               result,
               segmenter,
               reqFactory,
-              WriteFlushStrategy.fsyncEveryFlush(sc.writeObjectCallable()))) {
+              WriteFlushStrategy.fsyncEveryFlush(
+                  sc.writeObjectCallable(),
+                  RetryingDependencies.attemptOnce(),
+                  Retrying.neverRetry()))) {
         ImmutableList<ByteBuffer> buffers = TestUtils.subDivide(bytes, 10);
         for (ByteBuffer buf : buffers) {
           c.write(buf);
@@ -176,16 +192,113 @@ public final class GapicUnbufferedWritableByteChannelTest {
     }
   }
 
+  @Test
+  public void resumableUpload_chunkAutomaticRetry()
+      throws IOException, InterruptedException, ExecutionException {
+    AtomicBoolean req2SendErr = new AtomicBoolean(true);
+    AtomicBoolean req4SendErr = new AtomicBoolean(true);
+    AtomicInteger writeCount = new AtomicInteger(0);
+    StorageImplBase service =
+        new DirectWriteService(
+            (obs, requests) -> {
+              writeCount.getAndIncrement();
+              if (requests.equals(ImmutableList.of(req1))) {
+                obs.onNext(resp1);
+                obs.onCompleted();
+              } else if (requests.equals(ImmutableList.of(req2))) {
+                obs.onNext(resp2);
+                if (req2SendErr.get()) {
+                  req2SendErr.set(false);
+                  obs.onError(apiException(Code.DATA_LOSS));
+                } else {
+                  obs.onCompleted();
+                }
+              } else if (requests.equals(ImmutableList.of(req3))) {
+                obs.onNext(resp3);
+                obs.onCompleted();
+              } else if (requests.equals(ImmutableList.of(req4))) {
+                obs.onNext(resp4);
+                if (req4SendErr.get()) {
+                  req4SendErr.set(false);
+                  obs.onError(apiException(Code.DATA_LOSS));
+                } else {
+                  obs.onCompleted();
+                }
+              } else if (requests.equals(ImmutableList.of(req5))) {
+                obs.onNext(resp5);
+                obs.onCompleted();
+              } else {
+                obs.onError(
+                    TestUtils.apiException(Code.PERMISSION_DENIED, "Unexpected request chain."));
+              }
+            });
+    WriteCtx<?> writeCtx;
+    try (FakeServer fake = FakeServer.of(service);
+        StorageClient sc = StorageClient.create(fake.storageSettings())) {
+      SettableApiFuture<WriteObjectResponse> result = SettableApiFuture.create();
+      try (GapicUnbufferedWritableByteChannel<?> c =
+          new GapicUnbufferedWritableByteChannel<>(
+              result,
+              segmenter,
+              reqFactory,
+              WriteFlushStrategy.fsyncEveryFlush(
+                  sc.writeObjectCallable(),
+                  TestUtils.defaultRetryingDeps(),
+                  new BasicResultRetryAlgorithm<Object>() {
+                    @Override
+                    public boolean shouldRetry(Throwable t, Object ignore) {
+                      return t instanceof DataLossException;
+                    }
+                  }))) {
+        writeCtx = c.getWriteCtx();
+        ImmutableList<ByteBuffer> buffers = TestUtils.subDivide(bytes, 10);
+        c.write(buffers.get(0));
+        assertThat(writeCtx.getTotalSentBytes().get()).isEqualTo(10);
+        assertThat(writeCtx.getConfirmedBytes().get()).isEqualTo(10);
+        c.write(buffers.get(1));
+        assertThat(writeCtx.getTotalSentBytes().get()).isEqualTo(20);
+        assertThat(writeCtx.getConfirmedBytes().get()).isEqualTo(20);
+        c.write(buffers.get(2));
+        assertThat(writeCtx.getTotalSentBytes().get()).isEqualTo(30);
+        assertThat(writeCtx.getConfirmedBytes().get()).isEqualTo(30);
+        c.write(buffers.get(3));
+        assertThat(writeCtx.getTotalSentBytes().get()).isEqualTo(40);
+        assertThat(writeCtx.getConfirmedBytes().get()).isEqualTo(40);
+      }
+      assertThat(result.get()).isEqualTo(resp5);
+    }
+    assertThat(req2SendErr.get()).isFalse();
+    assertThat(req4SendErr.get()).isFalse();
+    assertThat(writeCount.get()).isEqualTo(7);
+
+    assertThat(writeCtx.getTotalSentBytes().get()).isEqualTo(40);
+    assertThat(writeCtx.getConfirmedBytes().get()).isEqualTo(40);
+  }
+
   private static class DirectWriteService extends StorageImplBase {
-    private final ImmutableMap<List<WriteObjectRequest>, WriteObjectResponse> writes;
+    private final BiConsumer<StreamObserver<WriteObjectResponse>, List<WriteObjectRequest>> c;
 
     private ImmutableList.Builder<WriteObjectRequest> requests;
 
-    private DirectWriteService(ImmutableMap<List<WriteObjectRequest>, WriteObjectResponse> writes) {
-      this.writes = writes;
+    private DirectWriteService(
+        BiConsumer<StreamObserver<WriteObjectResponse>, List<WriteObjectRequest>> c) {
+      this.c = c;
       this.requests = new ImmutableList.Builder<>();
     }
 
+    private DirectWriteService(ImmutableMap<List<WriteObjectRequest>, WriteObjectResponse> writes) {
+      this(
+          (obs, build) -> {
+            if (writes.containsKey(build)) {
+              obs.onNext(writes.get(build));
+              obs.onCompleted();
+            } else {
+              obs.onError(
+                  TestUtils.apiException(Code.PERMISSION_DENIED, "Unexpected request chain."));
+            }
+          });
+    }
+
     @Override
     public StreamObserver<WriteObjectRequest> writeObject(StreamObserver<WriteObjectResponse> obs) {
       return new Adapter() {
@@ -200,14 +313,8 @@ public final class GapicUnbufferedWritableByteChannelTest {
         @Override
         public void onCompleted() {
           ImmutableList<WriteObjectRequest> build = requests.build();
-          if (writes.containsKey(build)) {
-            requests = new ImmutableList.Builder<>();
-            obs.onNext(writes.get(build));
-            obs.onCompleted();
-          } else {
-            obs.onError(
-                TestUtils.apiException(Code.PERMISSION_DENIED, "Unexpected request chain."));
-          }
+          c.accept(obs, build);
+          requests = new ImmutableList.Builder<>();
         }
       };
     }
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/GrpcTransformPageDecoratorTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/GrpcTransformPageDecoratorTest.java
index 187348a9..f741ada0 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/GrpcTransformPageDecoratorTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/GrpcTransformPageDecoratorTest.java
@@ -18,20 +18,16 @@ package com.google.cloud.storage;
 
 import static com.google.common.truth.Truth.assertThat;
 
-import com.google.api.core.ApiClock;
 import com.google.api.core.ApiFuture;
 import com.google.api.core.ApiFutures;
-import com.google.api.core.NanoClock;
 import com.google.api.gax.paging.AbstractPage;
 import com.google.api.gax.retrying.ResultRetryAlgorithm;
-import com.google.api.gax.retrying.RetrySettings;
 import com.google.api.gax.retrying.TimedAttemptSettings;
 import com.google.api.gax.rpc.ApiCallContext;
 import com.google.api.gax.rpc.PageContext;
 import com.google.api.gax.rpc.PagedListDescriptor;
 import com.google.api.gax.rpc.UnaryCallable;
 import com.google.cloud.storage.GrpcStorageImpl.TransformingPageDecorator;
-import com.google.cloud.storage.Retrying.RetryingDependencies;
 import com.google.common.base.MoreObjects;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
@@ -85,7 +81,7 @@ public class GrpcTransformPageDecoratorTest {
         new TransformingPageDecorator<>(
             page,
             String::toUpperCase,
-            retryingDeps(),
+            TestUtils.defaultRetryingDeps(),
             StorageRetryStrategy.getUniformStorageRetryStrategy().getIdempotentHandler());
 
     assertThat(ImmutableList.copyOf(decorator.getValues().iterator()))
@@ -106,7 +102,8 @@ public class GrpcTransformPageDecoratorTest {
         PageContext.create(callable, descriptor, req1, apiCallContext);
     ReqRespPage page = new ReqRespPage(context, resp1);
     TransformingPageDecorator<Req, Resp, String, ReqRespPage, String> decorator =
-        new TransformingPageDecorator<>(page, String::toUpperCase, retryingDeps(), alg);
+        new TransformingPageDecorator<>(
+            page, String::toUpperCase, TestUtils.defaultRetryingDeps(), alg);
 
     ImmutableList<String> actual = ImmutableList.copyOf(decorator.iterateAll().iterator());
     assertThat(actual).containsExactlyElementsIn(expectedValues);
@@ -116,20 +113,6 @@ public class GrpcTransformPageDecoratorTest {
     assertThat(callable.callableCallCount.get()).isEqualTo(3);
   }
 
-  static RetryingDependencies retryingDeps() {
-    return new RetryingDependencies() {
-      @Override
-      public RetrySettings getRetrySettings() {
-        return StorageOptions.getDefaultRetrySettings();
-      }
-
-      @Override
-      public ApiClock getClock() {
-        return NanoClock.getDefaultClock();
-      }
-    };
-  }
-
   private static class ReqRespPage extends AbstractPage<Req, Resp, String, ReqRespPage> {
 
     private final Resp response;
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/TestUtils.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/TestUtils.java
index 573c2781..2bd1d54b 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/TestUtils.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/TestUtils.java
@@ -16,13 +16,17 @@
 
 package com.google.cloud.storage;
 
+import com.google.api.core.ApiClock;
+import com.google.api.core.NanoClock;
 import com.google.api.gax.grpc.GrpcCallContext;
 import com.google.api.gax.grpc.GrpcStatusCode;
+import com.google.api.gax.retrying.RetrySettings;
 import com.google.api.gax.rpc.ApiException;
 import com.google.api.gax.rpc.ApiExceptionFactory;
 import com.google.api.gax.rpc.ErrorDetails;
 import com.google.api.gax.rpc.StatusCode;
 import com.google.cloud.storage.Crc32cValue.Crc32cLengthKnown;
+import com.google.cloud.storage.Retrying.RetryingDependencies;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableSet;
 import com.google.protobuf.Any;
@@ -108,4 +112,18 @@ public final class TestUtils {
           .collect(ImmutableList.toImmutableList());
     }
   }
+
+  static RetryingDependencies defaultRetryingDeps() {
+    return new RetryingDependencies() {
+      @Override
+      public RetrySettings getRetrySettings() {
+        return StorageOptions.getDefaultRetrySettings();
+      }
+
+      @Override
+      public ApiClock getClock() {
+        return NanoClock.getDefaultClock();
+      }
+    };
+  }
 }
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/WriteFlushStrategyTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/WriteFlushStrategyTest.java
index 9ad61533..4c96d430 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/WriteFlushStrategyTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/WriteFlushStrategyTest.java
@@ -21,6 +21,7 @@ import static com.google.common.truth.Truth.assertThat;
 import com.google.api.gax.rpc.ApiCallContext;
 import com.google.api.gax.rpc.ApiStreamObserver;
 import com.google.api.gax.rpc.ClientStreamingCallable;
+import com.google.cloud.storage.Retrying.RetryingDependencies;
 import com.google.cloud.storage.WriteFlushStrategy.Flusher;
 import com.google.cloud.storage.WriteFlushStrategy.FlusherFactory;
 import com.google.common.collect.ImmutableList;
@@ -43,7 +44,12 @@ public final class WriteFlushStrategyTest {
 
   @Test
   public void bucketNameAddedToXGoogRequestParams_nonNull_nonEmpty_fsyncEveryFlush() {
-    doTest(WriteFlushStrategy::fsyncEveryFlush, "bucket-name", expectedHeaderNonNullNonEmpty);
+    doTest(
+        write ->
+            WriteFlushStrategy.fsyncEveryFlush(
+                write, RetryingDependencies.attemptOnce(), Retrying.neverRetry()),
+        "bucket-name",
+        expectedHeaderNonNullNonEmpty);
   }
 
   @Test
@@ -53,7 +59,12 @@ public final class WriteFlushStrategyTest {
 
   @Test
   public void bucketNameNotAddedToXGoogRequestParams_nonNull_empty_fsyncEveryFlush() {
-    doTest(WriteFlushStrategy::fsyncEveryFlush, "", expectedHeaderNonNullEmpty);
+    doTest(
+        write ->
+            WriteFlushStrategy.fsyncEveryFlush(
+                write, RetryingDependencies.attemptOnce(), Retrying.neverRetry()),
+        "",
+        expectedHeaderNonNullEmpty);
   }
 
   @Test
@@ -63,7 +74,12 @@ public final class WriteFlushStrategyTest {
 
   @Test
   public void bucketNameNotAddedToXGoogRequestParams_null_fsyncEveryFlush() {
-    doTest(WriteFlushStrategy::fsyncEveryFlush, null, expectedHeaderNull);
+    doTest(
+        write ->
+            WriteFlushStrategy.fsyncEveryFlush(
+                write, RetryingDependencies.attemptOnce(), Retrying.neverRetry()),
+        null,
+        expectedHeaderNull);
   }
 
   @Test