Repository: googleapis/java-storage
Pull request #1479 (2022-06-29 21:01:15)

chore: refactor GrpcStorageImpl#create* methods to use more specific channels

#### Direct Uploads
The semantics of flushing, closing, fsyncing and retrying are all different depending on
whether a direct or resumable upload is being used. GapicUnbufferedWritableByteChannel
doesn't really care about these semantics, instead it is responsible for converting
ByteBuffer[] into a number of WriteObjectRequest segments.

To encapsulate, and abstract away the semantic differences of upload types the new
Flusher class has been introduced. This new class provides two methods
1) `flush(@NonNull List<WriteObjectRequest>)`
2) `close(@Nullable WriteObjectRequest)`

With these two methods, GapicUnbufferedWritableByteChannel need not care about the
differences of direct and resumable uploads.

In the future, retry logic related to resumable and direct uploads can be centralized
into the respective Flusher directly along with the other semantics encapsulated there.

##### Flusher
Two implementations of Flusher are currently provided
1. FsyncEveryFlush - which will "fsync" for every invocation of flush() and close()
2. FsyncOnClose - which will only "fsync" when close() is called

#### Write Refactoring
1. Rename ServerState to WriteCtx.
   * This class is used more to track the information related to the processing of a
     WriteObject operation.
2. Move construction of the initial WriteObjectRequest.Builder into WriteCtx
   * The added level of indirection allows a single type to encapsulate both direct
     and resumable upload impacts on creating WriteObjectRequest.Builder instances
3. Delete GapicResumableUploadSessionBuilder.java and replaced with new
   GapicUploadSessionBuilder.java
4. Implement new type safe builder hierarchy for constructing direct upload sessions
   along with resumable
   * See GapicUploadSessionBuilderSyntaxTest.java for example usage and use cases
     the new builder hierarchy solves
6. Update GapicUnbufferedWritableByteChannel to leverage new WriteCtx and Flusher types

#### General Refactoring

1. Remove BufferHandle#allocateAligned in favor of alignment happening external to
   BufferHandle.
   * Move alignSize logic to Buffers and add Tests
2. Add Crc32cValue#debugString which can be used to generate a debug friendly string
   representation
4. Make close methods on Synchronized Channel decorators synchronized.
   * Most channel implementations have some work to do in their close method that is
     dependant upon state that happens during write
5. Fix null handling of null Crc32cValues when tracking cumulative for writes



Patch:
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/BufferHandle.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/BufferHandle.java
index d230e9d6..dfac4482 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/BufferHandle.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/BufferHandle.java
@@ -40,41 +40,14 @@ abstract class BufferHandle implements Supplier<ByteBuffer> {
 
   abstract int position();
 
-  static BufferHandle allocateAligned(int alignmentMultiple, int size) {
-    int actualSize = alignSize(alignmentMultiple, size);
-    return allocate(actualSize);
-  }
-
   static BufferHandle allocate(int capacity) {
-    return new LazyBufferHandle(capacity, ByteBuffer::allocate);
+    return new LazyBufferHandle(capacity, Buffers::allocate);
   }
 
   static BufferHandle handleOf(ByteBuffer buf) {
     return new EagerBufferHandle(buf);
   }
 
-  /**
-   * Give {@code size} "snap" it to the next {@code alignmentMultiple} that is >= {@code size}.
-   *
-   * <p>i.e. Given 344k size, 256k alignmentMultiple expect 512k
-   */
-  private static int alignSize(int alignmentMultiple, int size) {
-    int actualSize = size;
-    if (size < alignmentMultiple) {
-      actualSize = alignmentMultiple;
-    } else if (size % alignmentMultiple != 0) {
-      // TODO: this mod will cause two divisions to happen
-      //   * try and measure how expensive two divisions is compared to one
-      //   * also measure the case where size is a multiple, and how much the
-      //     following calculation costs
-
-      // add almost another full alignmentMultiple to the size
-      // then integer divide it before multiplying it by the alignmentMultiple
-      actualSize = (size + alignmentMultiple - 1) / alignmentMultiple * alignmentMultiple;
-    } // else size is already aligned
-    return actualSize;
-  }
-
   static final class LazyBufferHandle extends BufferHandle {
 
     private final int capacity;
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/Buffers.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/Buffers.java
index a39f254a..1ffce656 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/Buffers.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/Buffers.java
@@ -18,6 +18,7 @@ package com.google.cloud.storage;
 
 import java.nio.Buffer;
 import java.nio.ByteBuffer;
+import java.util.function.Consumer;
 
 /**
  * Utility methods for working with ByteBuffers
@@ -77,11 +78,7 @@ final class Buffers {
       if (bufRemaining == 0) {
         continue;
       } else if (bufRemaining < contentRemaining) {
-        ByteBuffer sub = content.duplicate();
-        int newLimit = sub.position() + bufRemaining;
-        sub.limit(newLimit);
-        buf.put(sub);
-        Buffers.position(content, newLimit);
+        sliceAndConsume(content, bufRemaining, buf::put);
       } else {
         buf.put(content);
       }
@@ -90,4 +87,45 @@ final class Buffers {
     }
     return total;
   }
+
+  /**
+   * Slice the provided source with a limit of {@code limit}, consume the slice with {@code c} then
+   * increment position of {@code src} to reflect the consumed bytes.
+   */
+  static void sliceAndConsume(ByteBuffer src, int limit, Consumer<ByteBuffer> c) {
+    ByteBuffer slice = src.slice();
+    slice.limit(limit);
+    c.accept(slice);
+    Buffers.position(src, src.position() + limit);
+  }
+
+  static ByteBuffer allocate(long l) {
+    return ByteBuffer.allocate(Math.toIntExact(l));
+  }
+
+  static ByteBuffer allocate(int i) {
+    return ByteBuffer.allocate(i);
+  }
+
+  /**
+   * Give {@code size} "snap" it to the next {@code alignmentMultiple} that is >= {@code size}.
+   *
+   * <p>i.e. Given 344k size, 256k alignmentMultiple expect 512k
+   */
+  static ByteBuffer allocateAligned(int size, int alignmentMultiple) {
+    int actualSize = size;
+    if (size < alignmentMultiple) {
+      actualSize = alignmentMultiple;
+    } else if (size % alignmentMultiple != 0) {
+      // TODO: this mod will cause two divisions to happen
+      //   * try and measure how expensive two divisions is compared to one
+      //   * also measure the case where size is a multiple, and how much the following calculation
+      //     costs
+
+      // add almost another full alignmentMultiple to the size
+      // then integer divide it before multiplying it by the alignmentMultiple
+      actualSize = (size + alignmentMultiple - 1) / alignmentMultiple * alignmentMultiple;
+    } // else size is already aligned
+    return allocate(actualSize);
+  }
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/Crc32cValue.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/Crc32cValue.java
index 1ab3c6bd..4062b507 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/Crc32cValue.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/Crc32cValue.java
@@ -47,6 +47,8 @@ abstract class Crc32cValue<Res extends Crc32cValue<Res>> {
    */
   public abstract Res concat(Crc32cLengthKnown other);
 
+  public abstract String debugString();
+
   static Crc32cLengthUnknown of(int value) {
     return new Crc32cLengthUnknown(value);
   }
@@ -55,6 +57,14 @@ abstract class Crc32cValue<Res extends Crc32cValue<Res>> {
     return new Crc32cLengthKnown(value, length);
   }
 
+  public static <Res extends Crc32cValue<Res>> Res nullSafeConcat(Res r1, Crc32cLengthKnown r2) {
+    if (r1 == null) {
+      return null;
+    } else {
+      return r1.concat(r2);
+    }
+  }
+
   static final class Crc32cLengthUnknown extends Crc32cValue<Crc32cLengthUnknown> {
     private final int value;
 
@@ -77,6 +87,11 @@ abstract class Crc32cValue<Res extends Crc32cValue<Res>> {
       return String.format("crc32c{0x%08x}", value);
     }
 
+    @Override
+    public String debugString() {
+      return toString();
+    }
+
     public Crc32cLengthKnown withLength(long length) {
       return new Crc32cLengthKnown(value, length);
     }
@@ -110,5 +125,10 @@ abstract class Crc32cValue<Res extends Crc32cValue<Res>> {
     public String toString() {
       return String.format("crc32c{0x%08x (length = %d)}", value, length);
     }
+
+    @Override
+    public String debugString() {
+      return String.format("crc32c{0x%08x}", value);
+    }
   }
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicDownloadSessionBuilder.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicDownloadSessionBuilder.java
index d371c5e1..6e7ea548 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicDownloadSessionBuilder.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicDownloadSessionBuilder.java
@@ -54,7 +54,7 @@ final class GapicDownloadSessionBuilder {
     }
 
     public BufferedReadableByteChannelSessionBuilder buffered() {
-      return buffered(ByteBuffer.allocate(16 * 1024 * 1024));
+      return buffered(Buffers.allocate(16 * 1024 * 1024));
     }
 
     public ReadableByteChannelSessionBuilder setHasher(Hasher hasher) {
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicResumableUploadSessionBuilder.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicResumableUploadSessionBuilder.java
deleted file mode 100644
index 62ff18db..00000000
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicResumableUploadSessionBuilder.java
+++ /dev/null
@@ -1,160 +0,0 @@
-/*
- * Copyright 2022 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 java.util.Objects.requireNonNull;
-
-import com.google.api.core.ApiFuture;
-import com.google.api.core.SettableApiFuture;
-import com.google.api.gax.rpc.ClientStreamingCallable;
-import com.google.cloud.storage.BufferedWritableByteChannelSession.BufferedWritableByteChannel;
-import com.google.cloud.storage.UnbufferedWritableByteChannelSession.UnbufferedWritableByteChannel;
-import com.google.storage.v2.WriteObjectRequest;
-import com.google.storage.v2.WriteObjectResponse;
-import java.nio.ByteBuffer;
-import java.util.function.BiFunction;
-
-final class GapicResumableUploadSessionBuilder {
-
-  private GapicResumableUploadSessionBuilder() {}
-
-  public static GapicResumableUploadSessionBuilder create() {
-    return new GapicResumableUploadSessionBuilder();
-  }
-
-  public WritableByteChannelSessionBuilder byteChannel(
-      ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write) {
-    return new WritableByteChannelSessionBuilder(write);
-  }
-
-  public static final class WritableByteChannelSessionBuilder {
-
-    private final ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write;
-    private ByteStringStrategy byteStringStrategy;
-    private Hasher hasher;
-
-    private WritableByteChannelSessionBuilder(
-        ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write) {
-      this.write = write;
-      this.byteStringStrategy = ByteStringStrategy.noCopy();
-      this.hasher = Hasher.noop();
-    }
-
-    public BufferedWritableByteChannelSessionBuilder buffered() {
-      return buffered(ByteBuffer.allocate(16 * 1024 * 1024));
-    }
-
-    public WritableByteChannelSessionBuilder setByteStringStrategy(
-        ByteStringStrategy byteStringStrategy) {
-      this.byteStringStrategy =
-          requireNonNull(byteStringStrategy, "checksumSupport must be non null");
-      return this;
-    }
-
-    public WritableByteChannelSessionBuilder setHasher(Hasher hasher) {
-      this.hasher = hasher;
-      return this;
-    }
-
-    public BufferedWritableByteChannelSessionBuilder buffered(ByteBuffer buffer) {
-      return new BufferedWritableByteChannelSessionBuilder(
-          BufferHandle.handleOf(buffer), getF(write, byteStringStrategy, hasher));
-    }
-
-    public BufferedWritableByteChannelSessionBuilder bufferedAndAligned(
-        int alignmentMultiple, int capacity) {
-      return new BufferedWritableByteChannelSessionBuilder(
-          BufferHandle.allocateAligned(alignmentMultiple, capacity),
-          getF(write, byteStringStrategy, hasher));
-    }
-
-    public UnbufferedWritableByteChannelSessionBuilder unbuffered() {
-      return new UnbufferedWritableByteChannelSessionBuilder(
-          getF(write, byteStringStrategy, hasher));
-    }
-
-    private static BiFunction<
-            ResumableWrite, SettableApiFuture<WriteObjectResponse>, UnbufferedWritableByteChannel>
-        getF(
-            ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write,
-            ByteStringStrategy byteStringStrategy,
-            Hasher hasher) {
-      return (start, resultFuture) ->
-          new GapicUnbufferedWritableByteChannel(
-              resultFuture, write, start, byteStringStrategy, hasher);
-    }
-
-    public static final class BufferedWritableByteChannelSessionBuilder {
-
-      private BiFunction<
-              ResumableWrite, SettableApiFuture<WriteObjectResponse>, BufferedWritableByteChannel>
-          f;
-      private ApiFuture<ResumableWrite> uploadIdFuture;
-
-      private BufferedWritableByteChannelSessionBuilder(
-          BufferHandle buffer,
-          BiFunction<
-                  ResumableWrite,
-                  SettableApiFuture<WriteObjectResponse>,
-                  UnbufferedWritableByteChannel>
-              f) {
-        this.f = f.andThen(c -> new DefaultBufferedWritableByteChannel(buffer, c));
-      }
-
-      public BufferedWritableByteChannelSessionBuilder setStartAsync(
-          ApiFuture<ResumableWrite> uploadIdFuture) {
-        this.uploadIdFuture = uploadIdFuture;
-        return this;
-      }
-
-      public BufferedWritableByteChannelSession<WriteObjectResponse> build() {
-        return new ChannelSession.BufferedWriteSession<>(
-            requireNonNull(uploadIdFuture, "uploadIdFuture must be non null"),
-            f.andThen(StorageByteChannels.writable()::createSynchronized));
-      }
-    }
-
-    public static final class UnbufferedWritableByteChannelSessionBuilder {
-
-      private BiFunction<
-              ResumableWrite, SettableApiFuture<WriteObjectResponse>, UnbufferedWritableByteChannel>
-          f;
-      private ApiFuture<ResumableWrite> uploadIdFuture;
-
-      private UnbufferedWritableByteChannelSessionBuilder(
-          BiFunction<
-                  ResumableWrite,
-                  SettableApiFuture<WriteObjectResponse>,
-                  UnbufferedWritableByteChannel>
-              f) {
-        this.f = f;
-      }
-
-      public UnbufferedWritableByteChannelSessionBuilder setStartAsync(
-          ApiFuture<ResumableWrite> uploadIdFuture) {
-        this.uploadIdFuture = uploadIdFuture;
-        return this;
-      }
-
-      public UnbufferedWritableByteChannelSession<WriteObjectResponse> build() {
-        return new ChannelSession.UnbufferedWriteSession<>(
-            requireNonNull(uploadIdFuture, "uploadIdFuture must be non null"),
-            f.andThen(StorageByteChannels.writable()::createSynchronized));
-      }
-    }
-  }
-}
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 890a35ce..c88a7221 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
@@ -17,12 +17,13 @@
 package com.google.cloud.storage;
 
 import com.google.api.core.SettableApiFuture;
-import com.google.api.gax.rpc.ApiStreamObserver;
 import com.google.api.gax.rpc.ClientStreamingCallable;
 import com.google.cloud.storage.ChunkSegmenter.ChunkSegment;
 import com.google.cloud.storage.Crc32cValue.Crc32cLengthKnown;
 import com.google.cloud.storage.UnbufferedWritableByteChannelSession.UnbufferedWritableByteChannel;
-import com.google.common.collect.ImmutableList;
+import com.google.cloud.storage.WriteCtx.WriteObjectRequestBuilderFactory;
+import com.google.cloud.storage.WriteFlushStrategy.Flusher;
+import com.google.cloud.storage.WriteFlushStrategy.FlusherFactory;
 import com.google.protobuf.ByteString;
 import com.google.storage.v2.ChecksummedData;
 import com.google.storage.v2.ObjectChecksums;
@@ -34,35 +35,37 @@ import java.nio.ByteBuffer;
 import java.nio.channels.ClosedChannelException;
 import java.util.ArrayList;
 import java.util.List;
-import java.util.concurrent.ExecutionException;
 
-final class GapicUnbufferedWritableByteChannel implements UnbufferedWritableByteChannel {
-
-  private final String uploadId;
+final class GapicUnbufferedWritableByteChannel<
+        RequestFactoryT extends WriteObjectRequestBuilderFactory>
+    implements UnbufferedWritableByteChannel {
 
   private final SettableApiFuture<WriteObjectResponse> resultFuture;
-  private final ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write;
   private final ByteStringStrategy byteStringStrategy;
   private final Hasher hasher;
-  private final ServerState serverState;
+  private final WriteCtx<RequestFactoryT> writeCtx;
   private final int perMessageLimit;
 
+  private final Flusher flusher;
+
   private boolean open = true;
   private boolean finished = false;
 
   GapicUnbufferedWritableByteChannel(
       SettableApiFuture<WriteObjectResponse> resultFuture,
       ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write,
-      ResumableWrite resumableWrite,
+      RequestFactoryT requestFactory,
       ByteStringStrategy byteStringStrategy,
-      Hasher hasher) {
-    this.uploadId = resumableWrite.getRes().getUploadId();
+      Hasher hasher,
+      FlusherFactory flusher) {
     this.resultFuture = resultFuture;
-    this.write = write; // todo: transform to retry once if request is non-idempotent
     this.byteStringStrategy = byteStringStrategy;
-    this.serverState = new ServerState(resumableWrite);
     this.hasher = hasher;
+
+    this.writeCtx = new WriteCtx<>(requestFactory);
     this.perMessageLimit = Values.MAX_WRITE_CHUNK_BYTES_VALUE;
+    this.flusher =
+        flusher.newFlusher(write, writeCtx.getConfirmedBytes()::addAndGet, resultFuture::set);
   }
 
   @Override
@@ -92,15 +95,15 @@ final class GapicUnbufferedWritableByteChannel implements UnbufferedWritableByte
       Crc32cLengthKnown crc32c = datum.getCrc32c();
       ByteString b = datum.getB();
       int contentSize = b.size();
-      long offset = serverState.getTotalSentBytes().getAndAdd(contentSize);
+      long offset = writeCtx.getTotalSentBytes().getAndAdd(contentSize);
       ChecksummedData.Builder checksummedData = ChecksummedData.newBuilder().setContent(b);
       if (crc32c != null) {
-        serverState.getCumulativeCrc32c().getAndAccumulate(crc32c, Crc32cValue::concat);
+        writeCtx.getCumulativeCrc32c().getAndAccumulate(crc32c, Crc32cValue::nullSafeConcat);
         checksummedData.setCrc32C(crc32c.getValue());
       }
       WriteObjectRequest.Builder builder =
-          WriteObjectRequest.newBuilder()
-              .setUploadId(uploadId)
+          writeCtx
+              .newRequestBuilder()
               .setWriteOffset(offset)
               .setChecksummedData(checksummedData.build());
       if (contentSize < perMessageLimit) {
@@ -108,7 +111,7 @@ final class GapicUnbufferedWritableByteChannel implements UnbufferedWritableByte
         if (crc32c != null) {
           builder.setObjectChecksums(
               ObjectChecksums.newBuilder()
-                  .setCrc32C(serverState.getCumulativeCrc32c().get().getValue())
+                  .setCrc32C(writeCtx.getCumulativeCrc32c().get().getValue())
                   .build());
         }
         finished = true;
@@ -119,7 +122,12 @@ final class GapicUnbufferedWritableByteChannel implements UnbufferedWritableByte
       bytesConsumed += contentSize;
     }
 
-    sendMessages(messages);
+    try {
+      flusher.flush(messages);
+    } catch (RuntimeException e) {
+      resultFuture.setException(e);
+      throw e;
+    }
 
     return bytesConsumed;
   }
@@ -132,83 +140,26 @@ final class GapicUnbufferedWritableByteChannel implements UnbufferedWritableByte
   @Override
   public void close() throws IOException {
     if (!finished) {
-      long offset = serverState.getTotalSentBytes().get();
+      long offset = writeCtx.getTotalSentBytes().get();
+      Crc32cLengthKnown crc32cValue = writeCtx.getCumulativeCrc32c().get();
+
       WriteObjectRequest.Builder b =
-          WriteObjectRequest.newBuilder()
-              .setUploadId(uploadId)
-              .setFinishWrite(true)
-              .setWriteOffset(offset);
-      Crc32cLengthKnown crc32cValue = serverState.getCumulativeCrc32c().get();
+          writeCtx.newRequestBuilder().setFinishWrite(true).setWriteOffset(offset);
       if (crc32cValue != null) {
         b.setObjectChecksums(
             ObjectChecksums.newBuilder().setCrc32C(crc32cValue.getValue()).build());
       }
       WriteObjectRequest message = b.build();
-      finished = true;
-      sendMessages(ImmutableList.of(message));
-    }
-    open = false;
-  }
-
-  private void sendMessages(List<WriteObjectRequest> messages) {
-
-    Observer observer = new Observer();
-    ApiStreamObserver<WriteObjectRequest> write = this.write.clientStreamingCall(observer);
-
-    messages.forEach(write::onNext);
-    write.onCompleted();
-    try {
-      observer.invocationHandle.get();
-    } catch (InterruptedException | ExecutionException e) {
-      if (e.getCause() instanceof RuntimeException) {
-        throw (RuntimeException) e.getCause();
-      } else {
-        throw new RuntimeException(e);
-      }
-    }
-  }
-
-  private class Observer implements ApiStreamObserver<WriteObjectResponse> {
-
-    private final SettableApiFuture<Void> invocationHandle;
-
-    private Observer() {
-      invocationHandle = SettableApiFuture.create();
-    }
-
-    @Override
-    public void onNext(WriteObjectResponse value) {
-      // incremental update
-      if (value.hasPersistedSize()) {
-        serverState.getConfirmedBytes().addAndGet(value.getPersistedSize());
-      } else if (value.hasResource()) {
-        // testbench_seems to return this even on finish
-      } else {
-        // serverState.confirmedBytes.set(serverState.totalSentBytes.get());
-      }
-      if (finished) {
-        resultFuture.set(value);
+      try {
+        flusher.close(message);
+        finished = true;
+      } catch (RuntimeException e) {
+        resultFuture.setException(e);
+        throw e;
       }
+    } else {
+      flusher.close(null);
     }
-
-    /**
-     * observed exceptions so far
-     *
-     * <ol>
-     *   <li>{@link com.google.api.gax.rpc.OutOfRangeException}
-     *   <li>{@link com.google.api.gax.rpc.AlreadyExistsException}
-     *   <li>{@link io.grpc.StatusRuntimeException}
-     * </ol>
-     */
-    @Override
-    public void onError(Throwable t) {
-      // TODO: is retryable?
-      invocationHandle.setException(t);
-    }
-
-    @Override
-    public void onCompleted() {
-      invocationHandle.set(null);
-    }
+    open = false;
   }
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUploadSessionBuilder.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUploadSessionBuilder.java
new file mode 100644
index 00000000..f1cc7c49
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicUploadSessionBuilder.java
@@ -0,0 +1,53 @@
+/*
+ * Copyright 2022 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.ApiFuture;
+import com.google.api.core.ApiFutures;
+import com.google.api.gax.rpc.ClientStreamingCallable;
+import com.google.api.gax.rpc.UnaryCallable;
+import com.google.common.util.concurrent.MoreExecutors;
+import com.google.storage.v2.StartResumableWriteRequest;
+import com.google.storage.v2.StartResumableWriteResponse;
+import com.google.storage.v2.WriteObjectRequest;
+import com.google.storage.v2.WriteObjectResponse;
+
+final class GapicUploadSessionBuilder {
+
+  private GapicUploadSessionBuilder() {}
+
+  static GapicUploadSessionBuilder create() {
+    return new GapicUploadSessionBuilder();
+  }
+
+  GapicWritableByteChannelSessionBuilder byteChannel(
+      ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write) {
+    return new GapicWritableByteChannelSessionBuilder(write);
+  }
+
+  ApiFuture<ResumableWrite> resumableWrite(
+      UnaryCallable<StartResumableWriteRequest, StartResumableWriteResponse> x,
+      WriteObjectRequest writeObjectRequest) {
+    StartResumableWriteRequest req =
+        StartResumableWriteRequest.newBuilder()
+            .setCommonObjectRequestParams(writeObjectRequest.getCommonObjectRequestParams())
+            .setWriteObjectSpec(writeObjectRequest.getWriteObjectSpec())
+            .build();
+    return ApiFutures.transform(
+        x.futureCall(req), (resp) -> new ResumableWrite(req, resp), MoreExecutors.directExecutor());
+  }
+}
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
new file mode 100644
index 00000000..553cda89
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicWritableByteChannelSessionBuilder.java
@@ -0,0 +1,292 @@
+/*
+ * Copyright 2022 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 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.rpc.ClientStreamingCallable;
+import com.google.cloud.storage.ChannelSession.BufferedWriteSession;
+import com.google.cloud.storage.ChannelSession.UnbufferedWriteSession;
+import com.google.cloud.storage.UnbufferedWritableByteChannelSession.UnbufferedWritableByteChannel;
+import com.google.cloud.storage.WriteCtx.WriteObjectRequestBuilderFactory;
+import com.google.cloud.storage.WriteFlushStrategy.FlusherFactory;
+import com.google.storage.v2.WriteObjectRequest;
+import com.google.storage.v2.WriteObjectResponse;
+import java.nio.ByteBuffer;
+import java.util.function.BiFunction;
+import java.util.function.Function;
+
+final class GapicWritableByteChannelSessionBuilder {
+
+  private static final int DEFAULT_BUFFER_CAPACITY = 16 * 1024 * 1024;
+  private final ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write;
+  private Hasher hasher;
+  private ByteStringStrategy byteStringStrategy;
+
+  GapicWritableByteChannelSessionBuilder(
+      ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write) {
+    this.write = write;
+    this.hasher = Hasher.noop();
+    this.byteStringStrategy = ByteStringStrategy.copy();
+  }
+
+  /**
+   * Set the {@link Hasher} to apply to the bytes passing through the built session's channel.
+   *
+   * <p>Default: {@link Hasher#noop()}
+   *
+   * @see Hasher#enabled()
+   * @see Hasher#noop()
+   */
+  GapicWritableByteChannelSessionBuilder setHasher(Hasher hasher) {
+    this.hasher = requireNonNull(hasher, "hasher must be non null");
+    return this;
+  }
+
+  /**
+   * Set the {@link ByteStringStrategy} to be used when constructing {@link
+   * com.google.protobuf.ByteString ByteString}s from {@link ByteBuffer}s.
+   *
+   * <p>Default: {@link ByteStringStrategy#copy()}
+   *
+   * <p>Note: usage of {@link ByteStringStrategy#noCopy()} requires that any {@link ByteBuffer}
+   * passed to the session's channel not be modified while {@link
+   * java.nio.channels.WritableByteChannel#write(ByteBuffer)} is processing.
+   *
+   * @see ByteStringStrategy#copy()
+   * @see ByteStringStrategy#noCopy()
+   */
+  GapicWritableByteChannelSessionBuilder setByteStringStrategy(
+      ByteStringStrategy byteStringStrategy) {
+    this.byteStringStrategy =
+        requireNonNull(byteStringStrategy, "byteStringStrategy must be non null");
+    return this;
+  }
+
+  /**
+   * The built {@link WritableByteChannelSession} will perform a "Direct" upload.
+   *
+   * <p>A "Direct" upload will only sync the transmitted data with GCS when the channel is closed.
+   * If an error is returned the entire upload will need to be attempted again.
+   */
+  DirectUploadBuilder direct() {
+    return new DirectUploadBuilder();
+  }
+
+  /**
+   * The build {@link WritableByteChannelSession} will perform a "Resumable" upload.
+   *
+   * <p>A "Resumable" upload will sync the transmitted data with GCS upon each individual flush and
+   * when the channel is closed.
+   *
+   * <p>If an error is returned the individual flush can be transparently retried.
+   */
+  ResumableUploadBuilder resumable() {
+    return new ResumableUploadBuilder();
+  }
+
+  /**
+   * When constructing any of our channel sessions, there is always a {@link
+   * GapicUnbufferedWritableByteChannel} at the bottom of it. This method creates a BiFunction which
+   * will instantiate the {@link GapicUnbufferedWritableByteChannel} when provided with a {@code
+   * StartT} value and a {@code SettableApiFuture<WriteObjectResponse>}.
+   *
+   * <p>As part of providing the function, the provided parameters {@code FlusherFactory} and {@code
+   * f} are "bound" into the returned function. In conjunction with the configured fields of this
+   * class a new instance of {@link GapicUnbufferedWritableByteChannel} can be constructed.
+   */
+  private <StartT, RequestFactoryT extends WriteObjectRequestBuilderFactory>
+      BiFunction<StartT, SettableApiFuture<WriteObjectResponse>, UnbufferedWritableByteChannel>
+          bindFunction(FlusherFactory flusherFactory, Function<StartT, RequestFactoryT> f) {
+    // it is theoretically possible that the setter methods for the following variables could
+    // be called again between when this method is invoked and the resulting function is invoked.
+    // To ensure we are using the specified values at the point in time they are bound to the
+    // function read them into local variables which will be closed over rather than the class
+    // fields.
+    ByteStringStrategy boundStrategy = byteStringStrategy;
+    Hasher boundHasher = hasher;
+    return (start, resultFuture) ->
+        new GapicUnbufferedWritableByteChannel<>(
+            resultFuture, write, f.apply(start), boundStrategy, boundHasher, flusherFactory);
+  }
+
+  final class DirectUploadBuilder {
+
+    /**
+     * Do not apply any intermediate buffering. Any call to {@link
+     * java.nio.channels.WritableByteChannel#write(ByteBuffer)} will be segmented as is and sent to
+     * GCS.
+     *
+     * <p>Note: this is considered an advanced API, and should not be used in circumstances in which
+     * control of {@link ByteBuffer}s sent to {@code write} is not self-contained.
+     */
+    UnbufferedDirectUploadBuilder unbuffered() {
+      return new UnbufferedDirectUploadBuilder();
+    }
+
+    /** Buffer up to {@link #DEFAULT_BUFFER_CAPACITY} worth of bytes before attempting to flush */
+    BufferedDirectUploadBuilder buffered() {
+      return buffered(BufferHandle.allocate(DEFAULT_BUFFER_CAPACITY));
+    }
+
+    /**
+     * Buffer using {@code byteBuffer} worth of space before attempting to flush.
+     *
+     * <p>The provided {@link ByteBuffer} <i>should</i> be aligned with GCSs block size of <a
+     * target="_blank" rel="noopener noreferrer"
+     * href="https://cloud.google.com/storage/docs/performing-resumable-uploads#chunked-upload">256
+     * KiB</a> if mu
+     */
+    BufferedDirectUploadBuilder buffered(ByteBuffer byteBuffer) {
+      return buffered(BufferHandle.handleOf(byteBuffer));
+    }
+
+    private BufferedDirectUploadBuilder buffered(BufferHandle bufferHandle) {
+      return new BufferedDirectUploadBuilder(bufferHandle);
+    }
+  }
+
+  final class ResumableUploadBuilder {
+
+    /**
+     * Do not apply any intermediate buffering. Any call to {@link
+     * java.nio.channels.WritableByteChannel#write(ByteBuffer)} will be segmented as is and sent to
+     * GCS.
+     *
+     * <p>Note: this is considered an advanced API, and should not be used in circumstances in which
+     * control of {@link ByteBuffer}s sent to {@code write} is not self-contained.
+     */
+    UnbufferedResumableUploadBuilder unbuffered() {
+      return new UnbufferedResumableUploadBuilder();
+    }
+
+    /** Buffer up to {@link #DEFAULT_BUFFER_CAPACITY} worth of bytes before attempting to flush */
+    BufferedResumableUploadBuilder buffered() {
+      return buffered(BufferHandle.allocate(DEFAULT_BUFFER_CAPACITY));
+    }
+
+    /**
+     * Buffer using {@code byteBuffer} worth of space before attempting to flush.
+     *
+     * <p>The provided {@link ByteBuffer} <i>should</i> be aligned with GCSs block size of <a
+     * target="_blank" rel="noopener noreferrer"
+     * href="https://cloud.google.com/storage/docs/performing-resumable-uploads#chunked-upload">256
+     * KiB</a>.
+     */
+    BufferedResumableUploadBuilder buffered(ByteBuffer byteBuffer) {
+      return buffered(BufferHandle.handleOf(byteBuffer));
+    }
+
+    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, 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, 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, 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;
+    }
+
+    BufferedWritableByteChannelSession<WriteObjectResponse> build() {
+      return new BufferedWriteSession<>(
+          requireNonNull(start, "start must be non null"),
+          bindFunction(WriteFlushStrategy::fsyncEveryFlush, 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/GrpcBlobReadChannel.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcBlobReadChannel.java
index d28f22c2..83910610 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcBlobReadChannel.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcBlobReadChannel.java
@@ -54,7 +54,7 @@ final class GrpcBlobReadChannel implements ReadChannel {
                       .read()
                       .byteChannel(read)
                       .setHasher(Hasher.noop())
-                      .buffered(ByteBuffer.allocate(chunkSize))
+                      .buffered(Buffers.allocate(chunkSize))
                       .setReadObjectRequest(req)
                       .build();
                 }));
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 5f42abd2..9774a076 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
@@ -16,6 +16,8 @@
 
 package com.google.cloud.storage;
 
+import static com.google.cloud.storage.GrpcStorageImpl._15MiB;
+import static com.google.cloud.storage.GrpcStorageImpl._256KiB;
 import static com.google.cloud.storage.Utils.todo;
 
 import com.google.api.core.ApiFuture;
@@ -35,7 +37,7 @@ final class GrpcBlobWriteChannel implements WriteChannel {
 
   private final LazyWriteChannel lazyWriteChannel;
 
-  private int chunkSize = 16 * 1024 * 1024;
+  private int chunkSize = _15MiB;
 
   GrpcBlobWriteChannel(
       ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write,
@@ -49,14 +51,14 @@ final class GrpcBlobWriteChannel implements WriteChannel {
                         .byteChannel(write)
                         .setHasher(Hasher.noop())
                         .setByteStringStrategy(ByteStringStrategy.copy())
-                        .buffered(ByteBuffer.allocate(chunkSize))
+                        .resumable()
+                        .buffered(Buffers.allocateAligned(chunkSize, _256KiB))
                         .setStartAsync(start.get())
                         .build()));
   }
 
   @Override
   public void setChunkSize(int chunkSize) {
-    // TODO: push value to next 256KiB boundary
     Preconditions.checkState(!isOpen(), "Unable to change chunkSize after write");
     this.chunkSize = chunkSize;
   }
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 38060a20..d4463770 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
@@ -22,7 +22,6 @@ import static com.google.common.base.MoreObjects.firstNonNull;
 import static java.util.Objects.requireNonNull;
 
 import com.google.api.core.ApiFuture;
-import com.google.api.core.ApiFutures;
 import com.google.api.gax.grpc.GrpcCallContext;
 import com.google.api.gax.paging.AbstractPage;
 import com.google.api.gax.paging.Page;
@@ -33,12 +32,14 @@ import com.google.cloud.BaseService;
 import com.google.cloud.Policy;
 import com.google.cloud.WriteChannel;
 import com.google.cloud.storage.Acl.Entity;
+import com.google.cloud.storage.BufferedWritableByteChannelSession.BufferedWritableByteChannel;
 import com.google.cloud.storage.Conversions.Decoder;
 import com.google.cloud.storage.HmacKey.HmacKeyMetadata;
 import com.google.cloud.storage.HmacKey.HmacKeyState;
 import com.google.cloud.storage.PostPolicyV4.PostConditionsV4;
 import com.google.cloud.storage.PostPolicyV4.PostFieldsV4;
 import com.google.cloud.storage.UnbufferedReadableByteChannelSession.UnbufferedReadableByteChannel;
+import com.google.cloud.storage.UnbufferedWritableByteChannelSession.UnbufferedWritableByteChannel;
 import com.google.cloud.storage.spi.v1.StorageRpc;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
@@ -46,7 +47,6 @@ import com.google.common.collect.ImmutableSet;
 import com.google.common.hash.HashCode;
 import com.google.common.hash.Hashing;
 import com.google.common.io.ByteStreams;
-import com.google.common.util.concurrent.MoreExecutors;
 import com.google.protobuf.ByteString;
 import com.google.storage.v2.BucketName;
 import com.google.storage.v2.CommonObjectRequestParams;
@@ -62,13 +62,13 @@ import com.google.storage.v2.ListHmacKeysRequest;
 import com.google.storage.v2.ListObjectsRequest;
 import com.google.storage.v2.Object;
 import com.google.storage.v2.ReadObjectRequest;
-import com.google.storage.v2.StartResumableWriteRequest;
 import com.google.storage.v2.StorageClient.ListBucketsPage;
 import com.google.storage.v2.StorageClient.ListBucketsPagedResponse;
 import com.google.storage.v2.StorageClient.ListHmacKeysPage;
 import com.google.storage.v2.StorageClient.ListHmacKeysPagedResponse;
 import com.google.storage.v2.StorageClient.ListObjectsPage;
 import com.google.storage.v2.StorageClient.ListObjectsPagedResponse;
+import com.google.storage.v2.WriteObjectRequest;
 import com.google.storage.v2.WriteObjectResponse;
 import com.google.storage.v2.WriteObjectSpec;
 import com.google.storage.v2.stub.GrpcStorageStub;
@@ -106,6 +106,8 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
           StandardOpenOption.WRITE,
           StandardOpenOption.CREATE,
           StandardOpenOption.TRUNCATE_EXISTING);
+  static final int _256KiB = 256 * 1024;
+  static final int _15MiB = 15 * 1024 * 1024;
 
   private final GrpcStorageStub grpcStorageStub;
   private final GrpcConversions codecs;
@@ -170,16 +172,27 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
       BlobInfo blobInfo, byte[] content, int offset, int length, BlobTargetOption... options) {
     requireNonNull(blobInfo, "blobInfo must be non null");
     requireNonNull(content, "content must be non null");
-    BlobWriteOption[] translate = translate(options);
+    Map<StorageRpc.Option, ?> optionsMap = StorageImpl.optionMap(options);
+    GrpcCallContext grpcCallContext = GrpcRequestMetadataSupport.create(optionsMap);
+    WriteObjectRequest req = getWriteObjectRequestBuilder(blobInfo, optionsMap).build();
     try {
-      ApiFuture<WriteObjectResponse> f;
-      try (GrpcBlobWriteChannel c = writer(blobInfo, translate)) {
+      UnbufferedWritableByteChannelSession<WriteObjectResponse> session =
+          ResumableMedia.gapic()
+              .write()
+              .byteChannel(
+                  grpcStorageStub.writeObjectCallable().withDefaultCallContext(grpcCallContext))
+              .setByteStringStrategy(ByteStringStrategy.noCopy())
+              .setHasher(Hasher.enabled())
+              .direct()
+              .unbuffered()
+              .setRequest(req)
+              .build();
+
+      try (UnbufferedWritableByteChannel c = session.open()) {
         c.write(ByteBuffer.wrap(content, offset, length));
-        f = c.getResults();
       }
-      WriteObjectResponse response = ApiExceptions.callAndTranslateApiException(f);
-      return syntaxDecoders.blob.decode(response.getResource());
-    } catch (IOException e) {
+      return getBlob(session.getResult());
+    } catch (Exception e) {
       throw StorageException.coalesce(e);
     }
   }
@@ -196,13 +209,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
   @Override
   public Blob createFrom(BlobInfo blobInfo, Path path, BlobWriteOption... options)
       throws IOException {
-    requireNonNull(path, "path must be non null");
-    if (Files.isDirectory(path)) {
-      throw new StorageException(0, path + " is a directory");
-    }
-    try (SeekableByteChannel src = Files.newByteChannel(path, READ_OPS)) {
-      return internalCreate(blobInfo, src, options);
-    }
+    return createFrom(blobInfo, path, _15MiB, options);
   }
 
   @Override
@@ -212,26 +219,101 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
     if (Files.isDirectory(path)) {
       throw new StorageException(0, path + " is a directory");
     }
-    try (SeekableByteChannel src = Files.newByteChannel(path, READ_OPS)) {
-      return internalCreate(blobInfo, src, bufferSize, options);
+
+    BlobTargetOption[] translate = translate(options);
+    // TODO: Why does optionMap not accept BlobWriteOption?
+    Map<StorageRpc.Option, ?> optionsMap = StorageImpl.optionMap(translate);
+    GrpcCallContext grpcCallContext = GrpcRequestMetadataSupport.create(optionsMap);
+    WriteObjectRequest req = getWriteObjectRequestBuilder(blobInfo, optionsMap).build();
+
+    GapicWritableByteChannelSessionBuilder channelSessionBuilder =
+        ResumableMedia.gapic()
+            .write()
+            .byteChannel(
+                grpcStorageStub.writeObjectCallable().withDefaultCallContext(grpcCallContext))
+            .setHasher(Hasher.enabled())
+            .setByteStringStrategy(ByteStringStrategy.noCopy());
+
+    BufferedWritableByteChannelSession<WriteObjectResponse> session;
+    long size = Files.size(path);
+    if (size < bufferSize) {
+      // ignore the bufferSize argument if the file is smaller than it
+      session =
+          channelSessionBuilder.direct().buffered(Buffers.allocate(size)).setRequest(req).build();
+    } else {
+      ApiFuture<ResumableWrite> start =
+          ResumableMedia.gapic()
+              .write()
+              .resumableWrite(
+                  grpcStorageStub
+                      .startResumableWriteCallable()
+                      .withDefaultCallContext(grpcCallContext),
+                  req);
+      session =
+          channelSessionBuilder
+              .resumable()
+              .buffered(Buffers.allocateAligned(bufferSize, _256KiB))
+              .setStartAsync(start)
+              .build();
     }
+
+    try (SeekableByteChannel src = Files.newByteChannel(path, READ_OPS);
+        BufferedWritableByteChannel dst = session.open()) {
+      ByteStreams.copy(src, dst);
+    } catch (Exception e) {
+      throw StorageException.coalesce(e);
+    }
+    return getBlob(session.getResult());
   }
 
   @Override
   public Blob createFrom(BlobInfo blobInfo, InputStream content, BlobWriteOption... options)
       throws IOException {
-    ReadableByteChannel src =
-        Channels.newChannel(firstNonNull(content, new ByteArrayInputStream(ZERO_BYTES)));
-    return internalCreate(blobInfo, src, options);
+    return createFrom(blobInfo, content, _15MiB, options);
   }
 
   @Override
   public Blob createFrom(
-      BlobInfo blobInfo, InputStream content, int bufferSize, BlobWriteOption... options)
+      BlobInfo blobInfo, InputStream in, int bufferSize, BlobWriteOption... options)
       throws IOException {
+    requireNonNull(blobInfo, "blobInfo must be non null");
+
+    BlobTargetOption[] translate = translate(options);
+    // TODO: Why does optionMap not accept BlobWriteOption?
+    Map<StorageRpc.Option, ?> optionsMap = StorageImpl.optionMap(translate);
+    GrpcCallContext grpcCallContext = GrpcRequestMetadataSupport.create(optionsMap);
+    WriteObjectRequest req = getWriteObjectRequestBuilder(blobInfo, optionsMap).build();
+
+    ApiFuture<ResumableWrite> start =
+        ResumableMedia.gapic()
+            .write()
+            .resumableWrite(
+                grpcStorageStub
+                    .startResumableWriteCallable()
+                    .withDefaultCallContext(grpcCallContext),
+                req);
+
+    BufferedWritableByteChannelSession<WriteObjectResponse> session =
+        ResumableMedia.gapic()
+            .write()
+            .byteChannel(
+                grpcStorageStub.writeObjectCallable().withDefaultCallContext(grpcCallContext))
+            .setHasher(Hasher.enabled())
+            .setByteStringStrategy(ByteStringStrategy.noCopy())
+            .resumable()
+            .buffered(Buffers.allocateAligned(bufferSize, _256KiB))
+            .setStartAsync(start)
+            .build();
+
+    // Specifically not in the try-with, so we don't close the provided stream
     ReadableByteChannel src =
-        Channels.newChannel(firstNonNull(content, new ByteArrayInputStream(ZERO_BYTES)));
-    return internalCreate(blobInfo, src, bufferSize, options);
+        Channels.newChannel(firstNonNull(in, new ByteArrayInputStream(ZERO_BYTES)));
+    try (BufferedWritableByteChannel dst = session.open()) {
+      ByteStreams.copy(src, dst);
+    } catch (Exception e) {
+      throw StorageException.coalesce(e);
+    }
+    return getBlob(session.getResult());
   }
 
   @Override
@@ -524,34 +606,21 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
 
   @Override
   public GrpcBlobWriteChannel writer(BlobInfo blobInfo, BlobWriteOption... options) {
-    Object object = codecs.blobInfo().encode(blobInfo);
-    WriteObjectSpec writeObjectSpec =
-        WriteObjectSpec.newBuilder()
-            .setResource(
-                object
-                    .toBuilder()
-                    // required if the data is changing
-                    .clearChecksums()
-                    // trimmed to shave payload size
-                    .clearAcl()
-                    .clearGeneration()
-                    .clearMetageneration()
-                    .clearSize()
-                    .clearCreateTime()
-                    .clearUpdateTime()
-                    .build())
-            .build();
-    StartResumableWriteRequest startResumableWriteRequest =
-        StartResumableWriteRequest.newBuilder().setWriteObjectSpec(writeObjectSpec).build();
+    BlobTargetOption[] translate = translate(options);
+    // TODO: Why does optionMap not accept BlobWriteOption?
+    Map<StorageRpc.Option, ?> optionsMap = StorageImpl.optionMap(translate);
+    GrpcCallContext grpcCallContext = GrpcRequestMetadataSupport.create(optionsMap);
+    WriteObjectRequest req = getWriteObjectRequestBuilder(blobInfo, optionsMap).build();
     return new GrpcBlobWriteChannel(
         grpcStorageStub.writeObjectCallable(),
         () ->
-            ApiFutures.transform(
-                grpcStorageStub
-                    .startResumableWriteCallable()
-                    .futureCall(startResumableWriteRequest),
-                (resp) -> new ResumableWrite(startResumableWriteRequest, resp),
-                MoreExecutors.directExecutor()));
+            ResumableMedia.gapic()
+                .write()
+                .resumableWrite(
+                    grpcStorageStub
+                        .startResumableWriteCallable()
+                        .withDefaultCallContext(grpcCallContext),
+                    req));
   }
 
   @Override
@@ -875,29 +944,13 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
     return todo();
   }
 
-  private Blob internalCreate(
-      BlobInfo blobInfo, ReadableByteChannel src, BlobWriteOption... options) throws IOException {
-    requireNonNull(blobInfo, "blobInfo must be non null");
-    ApiFuture<WriteObjectResponse> f;
-    try (GrpcBlobWriteChannel c = writer(blobInfo, options)) {
-      ByteStreams.copy(src, c);
-      f = c.getResults();
-    }
-    WriteObjectResponse response = ApiExceptions.callAndTranslateApiException(f);
-    return syntaxDecoders.blob.decode(response.getResource());
+  private BlobTargetOption[] translate(BlobWriteOption[] options) {
+    return todo();
   }
 
-  private Blob internalCreate(
-      BlobInfo blobInfo, ReadableByteChannel src, int bufferSize, BlobWriteOption... options)
-      throws IOException {
-    requireNonNull(blobInfo, "blobInfo must be non null");
-    ApiFuture<WriteObjectResponse> f;
-    try (GrpcBlobWriteChannel c = writer(blobInfo, options)) {
-      c.setChunkSize(bufferSize);
-      ByteStreams.copy(src, c);
-      f = c.getResults();
-    }
-    WriteObjectResponse response = ApiExceptions.callAndTranslateApiException(f);
+  private Blob getBlob(ApiFuture<WriteObjectResponse> result) {
+    // TODO: investigate if we need to unnest any exception
+    WriteObjectResponse response = ApiExceptions.callAndTranslateApiException(result);
     return syntaxDecoders.blob.decode(response.getResource());
   }
 
@@ -1018,13 +1071,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
         builder::setIfGenerationNotMatch);
     String key = (String) optionsMap.get(StorageRpc.Option.CUSTOMER_SUPPLIED_KEY);
     if (key != null) {
-      CommonObjectRequestParams.Builder paramsBuilder = CommonObjectRequestParams.newBuilder();
-      paramsBuilder.setEncryptionAlgorithm("AES256");
-      paramsBuilder.setEncryptionKeyBytes(ByteString.copyFromUtf8(key));
-      byte[] keyBytes = Base64.getDecoder().decode(key);
-      HashCode keySha256 = Hashing.sha256().hashBytes(keyBytes);
-      paramsBuilder.setEncryptionKeySha256Bytes(ByteString.copyFrom(keySha256.asBytes()));
-      builder.setCommonObjectRequestParams(paramsBuilder.build());
+      builder.setCommonObjectRequestParams(commonRequestParams(key));
     }
     return builder.build();
   }
@@ -1044,4 +1091,67 @@ final class GrpcStorageImpl extends BaseService<StorageOptions> implements Stora
       return ctx;
     }
   }
+
+  private WriteObjectRequest.Builder getWriteObjectRequestBuilder(
+      BlobInfo info, Map<StorageRpc.Option, ?> optionsMap) {
+    Object object = codecs.blobInfo().encode(info);
+    Object.Builder objectBuilder =
+        object
+            .toBuilder()
+            // required if the data is changing
+            .clearChecksums()
+            // trimmed to shave payload size
+            .clearAcl()
+            .clearGeneration()
+            .clearMetageneration()
+            .clearSize()
+            .clearCreateTime()
+            .clearUpdateTime();
+    WriteObjectSpec.Builder specBuilder = WriteObjectSpec.newBuilder().setResource(objectBuilder);
+
+    WriteObjectRequest.Builder requestBuilder =
+        WriteObjectRequest.newBuilder().setWriteObjectSpec(specBuilder);
+
+    // TODO: Projection: Do we care?
+
+    ifNonNull(
+        (String) optionsMap.get(StorageRpc.Option.PREDEFINED_ACL), specBuilder::setPredefinedAcl);
+    ifNonNull(
+        (Long) optionsMap.get(StorageRpc.Option.IF_METAGENERATION_MATCH),
+        specBuilder::setIfMetagenerationMatch);
+    ifNonNull(
+        (Long) optionsMap.get(StorageRpc.Option.IF_METAGENERATION_NOT_MATCH),
+        specBuilder::setIfMetagenerationNotMatch);
+    ifNonNull(
+        (Long) optionsMap.get(StorageRpc.Option.IF_GENERATION_MATCH),
+        specBuilder::setIfGenerationMatch);
+    ifNonNull(
+        (Long) optionsMap.get(StorageRpc.Option.IF_GENERATION_NOT_MATCH),
+        specBuilder::setIfGenerationNotMatch);
+    String userProject = (String) optionsMap.get(StorageRpc.Option.USER_PROJECT);
+    if (userProject != null) {
+      // TODO: user project
+    }
+    String encryptionKey = (String) optionsMap.get(StorageRpc.Option.CUSTOMER_SUPPLIED_KEY);
+    if (encryptionKey != null) {
+      requestBuilder.setCommonObjectRequestParams(commonRequestParams(encryptionKey));
+    }
+    String key = (String) optionsMap.get(StorageRpc.Option.KMS_KEY_NAME);
+    if (key != null) {
+      objectBuilder.setKmsKey(key);
+    }
+
+    return requestBuilder;
+  }
+
+  private CommonObjectRequestParams commonRequestParams(String key) {
+    byte[] keyBytes = Base64.getDecoder().decode(key);
+    HashCode keySha256 = Hashing.sha256().hashBytes(keyBytes);
+
+    return CommonObjectRequestParams.newBuilder()
+        .setEncryptionAlgorithm("AES256")
+        .setEncryptionKeyBytes(ByteString.copyFromUtf8(key))
+        .setEncryptionKeySha256Bytes(ByteString.copyFrom(keySha256.asBytes()))
+        .build();
+  }
 }
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
index dcc17f6a..caf2e44f 100644
--- 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
@@ -85,8 +85,8 @@ final class ResumableMedia {
 
     private GapicMediaSession() {}
 
-    GapicResumableUploadSessionBuilder write() {
-      return GapicResumableUploadSessionBuilder.create();
+    GapicUploadSessionBuilder write() {
+      return GapicUploadSessionBuilder.create();
     }
 
     GapicDownloadSessionBuilder read() {
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/ResumableWrite.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/ResumableWrite.java
index 030def9a..17e599e0 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/ResumableWrite.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/ResumableWrite.java
@@ -16,18 +16,31 @@
 
 package com.google.cloud.storage;
 
+import static com.google.cloud.storage.StorageV2ProtoUtils.fmtProto;
+
+import com.google.cloud.storage.WriteCtx.WriteObjectRequestBuilderFactory;
 import com.google.storage.v2.StartResumableWriteRequest;
 import com.google.storage.v2.StartResumableWriteResponse;
+import com.google.storage.v2.WriteObjectRequest;
 import java.util.Objects;
+import java.util.function.Function;
 
-final class ResumableWrite {
+final class ResumableWrite implements WriteObjectRequestBuilderFactory {
 
   private final StartResumableWriteRequest req;
   private final StartResumableWriteResponse res;
 
+  private final WriteObjectRequest writeRequest;
+
   public ResumableWrite(StartResumableWriteRequest req, StartResumableWriteResponse res) {
     this.req = req;
     this.res = res;
+    this.writeRequest =
+        WriteObjectRequest.newBuilder()
+            .setUploadId(res.getUploadId())
+            .setCommonObjectRequestParams(req.getCommonObjectRequestParams())
+            // REVIEW: how does kms come into play here?
+            .build();
   }
 
   public StartResumableWriteRequest getReq() {
@@ -38,9 +51,14 @@ final class ResumableWrite {
     return res;
   }
 
+  @Override
+  public WriteObjectRequest.Builder newBuilder() {
+    return writeRequest.toBuilder();
+  }
+
   @Override
   public String toString() {
-    return "Start{" + "req=" + req + ", res=" + res + '}';
+    return "ResumableWrite{" + "req=" + fmtProto(req) + ", res=" + fmtProto(res) + '}';
   }
 
   @Override
@@ -59,4 +77,12 @@ final class ResumableWrite {
   public int hashCode() {
     return Objects.hash(req, res);
   }
+
+  /**
+   * Helper function which is more specific than {@link Function#identity()}. Constraining the input
+   * and output to be exactly {@link ResumableWrite}.
+   */
+  static ResumableWrite identity(ResumableWrite w) {
+    return w;
+  }
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageByteChannels.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageByteChannels.java
index d0898b76..cd4410db 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageByteChannels.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageByteChannels.java
@@ -83,7 +83,7 @@ final class StorageByteChannels {
     }
 
     @Override
-    public void close() throws IOException {
+    public synchronized void close() throws IOException {
       delegate.close();
     }
   }
@@ -108,7 +108,7 @@ final class StorageByteChannels {
     }
 
     @Override
-    public void close() throws IOException {
+    public synchronized void close() throws IOException {
       delegate.close();
     }
 
@@ -148,7 +148,7 @@ final class StorageByteChannels {
     }
 
     @Override
-    public void close() throws IOException {
+    public synchronized void close() throws IOException {
       delegate.close();
     }
   }
@@ -183,7 +183,7 @@ final class StorageByteChannels {
     }
 
     @Override
-    public void close() throws IOException {
+    public synchronized void close() throws IOException {
       delegate.close();
     }
   }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/ServerState.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/WriteCtx.java
similarity index 51%
rename from google-cloud-storage/src/main/java/com/google/cloud/storage/ServerState.java
rename to google-cloud-storage/src/main/java/com/google/cloud/storage/WriteCtx.java
index 3732f626..9e55d8cc 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/ServerState.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/WriteCtx.java
@@ -16,26 +16,31 @@
 
 package com.google.cloud.storage;
 
+import static com.google.cloud.storage.StorageV2ProtoUtils.fmtProto;
+
 import com.google.cloud.storage.Crc32cValue.Crc32cLengthKnown;
+import com.google.cloud.storage.WriteCtx.WriteObjectRequestBuilderFactory;
+import com.google.storage.v2.WriteObjectRequest;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
 
-final class ServerState {
+final class WriteCtx<RequestFactoryT extends WriteObjectRequestBuilderFactory> {
+
+  private final RequestFactoryT requestFactory;
 
-  private final ResumableWrite resumableWrite;
   private final AtomicLong totalSentBytes;
   private final AtomicLong confirmedBytes;
   private final AtomicReference<Crc32cLengthKnown> cumulativeCrc32c;
 
-  public ServerState(ResumableWrite resumableWrite) {
-    this.resumableWrite = resumableWrite;
-    totalSentBytes = new AtomicLong(0);
-    confirmedBytes = new AtomicLong(0);
-    cumulativeCrc32c = new AtomicReference<>();
+  WriteCtx(RequestFactoryT requestFactory) {
+    this.requestFactory = requestFactory;
+    this.totalSentBytes = new AtomicLong(0);
+    this.confirmedBytes = new AtomicLong(0);
+    this.cumulativeCrc32c = new AtomicReference<>();
   }
 
-  public ResumableWrite getStart() {
-    return resumableWrite;
+  public WriteObjectRequest.Builder newRequestBuilder() {
+    return requestFactory.newBuilder();
   }
 
   public AtomicLong getTotalSentBytes() {
@@ -58,8 +63,8 @@ final class ServerState {
   @Override
   public String toString() {
     return "ServerState{"
-        + "start="
-        + resumableWrite
+        + "requestFactory="
+        + requestFactory
         + ", totalSentBytes="
         + totalSentBytes
         + ", confirmedBytes="
@@ -68,4 +73,32 @@ final class ServerState {
         + cumulativeCrc32c
         + '}';
   }
+
+  @FunctionalInterface
+  interface WriteObjectRequestBuilderFactory {
+    WriteObjectRequest.Builder newBuilder();
+
+    static SimpleWriteObjectRequestBuilderFactory simple(WriteObjectRequest req) {
+      return new SimpleWriteObjectRequestBuilderFactory(req);
+    }
+  }
+
+  static final class SimpleWriteObjectRequestBuilderFactory
+      implements WriteObjectRequestBuilderFactory {
+    private final WriteObjectRequest req;
+
+    private SimpleWriteObjectRequestBuilderFactory(WriteObjectRequest req) {
+      this.req = req;
+    }
+
+    @Override
+    public WriteObjectRequest.Builder newBuilder() {
+      return req.toBuilder();
+    }
+
+    @Override
+    public String toString() {
+      return "SimpleWriteObjectRequestBuilderFactory{" + "req=" + fmtProto(req) + '}';
+    }
+  }
 }
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
new file mode 100644
index 00000000..1104c18b
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/WriteFlushStrategy.java
@@ -0,0 +1,246 @@
+/*
+ * Copyright 2022 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.SettableApiFuture;
+import com.google.api.gax.rpc.ApiStreamObserver;
+import com.google.api.gax.rpc.ClientStreamingCallable;
+import com.google.common.collect.ImmutableList;
+import com.google.storage.v2.WriteObjectRequest;
+import com.google.storage.v2.WriteObjectResponse;
+import java.util.List;
+import java.util.concurrent.ExecutionException;
+import java.util.function.Consumer;
+import java.util.function.LongConsumer;
+import org.checkerframework.checker.nullness.qual.NonNull;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+/**
+ * When writing to GCS using the WriteObject rpc, there are some behavioral differences between
+ * performing a "direct" upload vs a "resumable" upload.
+ *
+ * <p>This class provides the encapsulation of the stream and "fsync" lifecycles and possible
+ * automatic retry where applicable.
+ *
+ * <p>In this class "fsync" is used to mean "complete the client stream to GCS and await its
+ * response". We are loosely following the concept used in linux to flush data to disk <a
+ * target="_blank" rel="noopener noreferrer"
+ * href="https://man7.org/linux/man-pages/man2/fdatasync.2.html">fsync(2)</a>
+ */
+final class WriteFlushStrategy {
+
+  private WriteFlushStrategy() {}
+
+  /**
+   * Create a {@link Flusher} which will "fsync" every time {@link Flusher#flush(List)} is called
+   * along with {@link Flusher#close(WriteObjectRequest)}.
+   */
+  static Flusher fsyncEveryFlush(
+      ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write,
+      LongConsumer committedTotalBytesCallback,
+      Consumer<WriteObjectResponse> onSuccessCallback) {
+    return new FsyncEveryFlusher(write, committedTotalBytesCallback, onSuccessCallback);
+  }
+
+  /**
+   * Create a {@link Flusher} which will "fsync" only on {@link Flusher#close(WriteObjectRequest)}.
+   * Calls to {@link Flusher#flush(List)} will be sent but not synced.
+   *
+   * @see FlusherFactory#newFlusher(ClientStreamingCallable, LongConsumer, Consumer)
+   */
+  static Flusher fsyncOnClose(
+      ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write,
+      LongConsumer committedTotalBytesCallback,
+      Consumer<WriteObjectResponse> onSuccessCallback) {
+    return new FsyncOnClose(write, committedTotalBytesCallback, onSuccessCallback);
+  }
+
+  @FunctionalInterface
+  interface FlusherFactory {
+    /**
+     * @param committedTotalBytesCallback Callback to signal the total number of bytes committed by
+     *     this flusher.
+     * @param onSuccessCallback Callback to signal success, and provide the final response.
+     */
+    Flusher newFlusher(
+        ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write,
+        LongConsumer committedTotalBytesCallback,
+        Consumer<WriteObjectResponse> onSuccessCallback);
+  }
+
+  interface Flusher {
+    void flush(@NonNull List<WriteObjectRequest> segments);
+
+    void close(@Nullable WriteObjectRequest req);
+  }
+
+  private static final class FsyncEveryFlusher implements Flusher {
+
+    private final ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write;
+    private final LongConsumer sizeCallback;
+    private final Consumer<WriteObjectResponse> completeCallback;
+
+    private FsyncEveryFlusher(
+        ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write,
+        LongConsumer sizeCallback,
+        Consumer<WriteObjectResponse> completeCallback) {
+      this.write = write;
+      this.sizeCallback = sizeCallback;
+      this.completeCallback = completeCallback;
+    }
+
+    public void flush(@NonNull List<WriteObjectRequest> segments) {
+
+      Observer observer = new Observer(sizeCallback, completeCallback);
+      ApiStreamObserver<WriteObjectRequest> write = this.write.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();
+    }
+
+    public void close(@Nullable WriteObjectRequest req) {
+      if (req != null) {
+        flush(ImmutableList.of(req));
+      }
+    }
+  }
+
+  private static final class FsyncOnClose implements Flusher {
+
+    private final ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write;
+    private final Observer responseObserver;
+
+    private volatile ApiStreamObserver<WriteObjectRequest> stream;
+    private boolean first = true;
+
+    private FsyncOnClose(
+        ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write,
+        LongConsumer sizeCallback,
+        Consumer<WriteObjectResponse> completeCallback) {
+      this.write = write;
+      this.responseObserver = new Observer(sizeCallback, completeCallback);
+    }
+
+    @Override
+    public void flush(@NonNull List<WriteObjectRequest> segments) {
+      ensureOpen();
+      for (WriteObjectRequest message : segments) {
+        if (!first) {
+          message = message.toBuilder().clearUploadId().clearWriteObjectSpec().build();
+        }
+
+        stream.onNext(message);
+        first = false;
+      }
+    }
+
+    @Override
+    public void close(@Nullable WriteObjectRequest message) {
+      ensureOpen();
+      if (message != null) {
+        if (!first) {
+          message = message.toBuilder().clearUploadId().clearWriteObjectSpec().build();
+        }
+        stream.onNext(message);
+      }
+      stream.onCompleted();
+      responseObserver.await();
+    }
+
+    private void ensureOpen() {
+      if (stream == null) {
+        synchronized (this) {
+          if (stream == null) {
+            stream = write.clientStreamingCall(responseObserver);
+          }
+        }
+      }
+    }
+  }
+
+  private static class Observer implements ApiStreamObserver<WriteObjectResponse> {
+
+    private final LongConsumer sizeCallback;
+    private final Consumer<WriteObjectResponse> completeCallback;
+
+    private final SettableApiFuture<Void> invocationHandle;
+    private volatile WriteObjectResponse last;
+
+    private Observer(LongConsumer sizeCallback, Consumer<WriteObjectResponse> completeCallback) {
+      this.sizeCallback = sizeCallback;
+      this.completeCallback = completeCallback;
+      this.invocationHandle = SettableApiFuture.create();
+    }
+
+    @Override
+    public void onNext(WriteObjectResponse value) {
+      // incremental update
+      if (value.hasPersistedSize()) {
+        sizeCallback.accept(value.getPersistedSize());
+      } else if (value.hasResource()) {
+        // testbench_seems to return this even on finish
+      } else {
+        // serverState.confirmedBytes.set(serverState.totalSentBytes.get());
+      }
+      last = value;
+    }
+
+    /**
+     * observed exceptions so far
+     *
+     * <ol>
+     *   <li>{@link com.google.api.gax.rpc.OutOfRangeException}
+     *   <li>{@link com.google.api.gax.rpc.AlreadyExistsException}
+     *   <li>{@link io.grpc.StatusRuntimeException}
+     * </ol>
+     */
+    @Override
+    public void onError(Throwable t) {
+      // TODO: is retryable?
+      invocationHandle.setException(t);
+    }
+
+    @Override
+    public void onCompleted() {
+      if (last != null && last.hasResource()) {
+        completeCallback.accept(last);
+      }
+      invocationHandle.set(null);
+    }
+
+    private void await() {
+      try {
+        invocationHandle.get();
+      } catch (InterruptedException | ExecutionException e) {
+        if (e.getCause() instanceof RuntimeException) {
+          throw (RuntimeException) e.getCause();
+        } else {
+          throw new RuntimeException(e);
+        }
+      }
+    }
+  }
+}


Test patch:
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/BuffersTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/BuffersTest.java
index 926db586..f0d760cb 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/BuffersTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/BuffersTest.java
@@ -20,12 +20,12 @@ import static com.google.common.truth.Truth.assertThat;
 
 import java.nio.ByteBuffer;
 import java.security.SecureRandom;
-import net.jqwik.api.Example;
+import org.junit.Test;
 
 public final class BuffersTest {
 
-  @Example
-  void copy() {
+  @Test
+  public void copy() {
     SecureRandom rand = new SecureRandom();
     ByteBuffer content = DataGenerator.rand(rand).genByteBuffer(2048);
 
@@ -48,4 +48,28 @@ public final class BuffersTest {
     }
     assertThat(copy).isEqualTo(2048);
   }
+
+  @Test
+  public void allocateAligned_nonDivisible_capacityGtAlignment() {
+    ByteBuffer b1 = Buffers.allocateAligned(3, 2);
+    assertThat(b1.capacity()).isEqualTo(4);
+  }
+
+  @Test
+  public void allocateAligned_nonDivisible_capacityLtAlignment() {
+    ByteBuffer b1 = Buffers.allocateAligned(1, 2);
+    assertThat(b1.capacity()).isEqualTo(2);
+  }
+
+  @Test
+  public void allocateAligned_evenlyDivisible_capacityLtAlignment() {
+    ByteBuffer b1 = Buffers.allocateAligned(2, 4);
+    assertThat(b1.capacity()).isEqualTo(4);
+  }
+
+  @Test
+  public void allocateAligned_evenlyDivisible_capacityGtAlignment() {
+    ByteBuffer b1 = Buffers.allocateAligned(8, 4);
+    assertThat(b1.capacity()).isEqualTo(8);
+  }
 }
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/GapicUploadSessionBuilderSyntaxTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/GapicUploadSessionBuilderSyntaxTest.java
new file mode 100644
index 00000000..a8afe473
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/GapicUploadSessionBuilderSyntaxTest.java
@@ -0,0 +1,177 @@
+/*
+ * Copyright 2022 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.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.when;
+
+import com.google.api.core.ApiFuture;
+import com.google.api.core.ApiFutures;
+import com.google.api.gax.rpc.ClientStreamingCallable;
+import com.google.api.gax.rpc.UnaryCallable;
+import com.google.storage.v2.StartResumableWriteRequest;
+import com.google.storage.v2.StartResumableWriteResponse;
+import com.google.storage.v2.WriteObjectRequest;
+import com.google.storage.v2.WriteObjectResponse;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+
+/**
+ * For our {@link GapicUploadSessionBuilder} a number of items must be provided to ultimately
+ * construct an instance. This class enumerates some supported usage patterns whether using the
+ * fluent api from beginning to build or if intermediate variables must be defined along the way.
+ *
+ * <p>These "tests" are primarily validated at compilation time - ensuring source compatibility of
+ * the builders. Each builder is built which runs some minor validation - such as null checks - but
+ * otherwise does not have any other logic.
+ */
+@RunWith(MockitoJUnitRunner.class)
+public final class GapicUploadSessionBuilderSyntaxTest {
+
+  private final WriteObjectRequest req = WriteObjectRequest.getDefaultInstance();
+
+  // The following fields are "mocks" for simplicity’s sake.
+  // We need them to be non-null, but otherwise they do not matter.
+  // They have many dependencies which would need to be constructed in order to instantiate a
+  // literal instance.
+  @Mock private ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write;
+
+  @Mock
+  private UnaryCallable<StartResumableWriteRequest, StartResumableWriteResponse>
+      startResumableWrite;
+
+  @Before
+  public void setUp() throws Exception {
+    when(startResumableWrite.futureCall(any()))
+        .thenReturn(ApiFutures.immediateFuture(StartResumableWriteResponse.getDefaultInstance()));
+  }
+
+  @Test
+  public void syntax_directUnbuffered_fluent() {
+    UnbufferedWritableByteChannelSession<WriteObjectResponse> session =
+        ResumableMedia.gapic()
+            .write()
+            .byteChannel(write)
+            .setHasher(Hasher.noop())
+            .setByteStringStrategy(ByteStringStrategy.noCopy())
+            .direct()
+            .unbuffered()
+            .setRequest(req)
+            .build();
+  }
+
+  @Test
+  public void syntax_directBuffered_fluent() {
+    BufferedWritableByteChannelSession<WriteObjectResponse> session =
+        ResumableMedia.gapic()
+            .write()
+            .byteChannel(write)
+            .setHasher(Hasher.noop())
+            .setByteStringStrategy(ByteStringStrategy.noCopy())
+            .direct()
+            .buffered()
+            .setRequest(req)
+            .build();
+  }
+
+  @Test
+  public void syntax_resumableUnbuffered_fluent() {
+    ApiFuture<ResumableWrite> startAsync =
+        ResumableMedia.gapic().write().resumableWrite(startResumableWrite, req);
+    UnbufferedWritableByteChannelSession<WriteObjectResponse> session =
+        ResumableMedia.gapic()
+            .write()
+            .byteChannel(write)
+            .setHasher(Hasher.noop())
+            .setByteStringStrategy(ByteStringStrategy.noCopy())
+            .resumable()
+            .unbuffered()
+            .setStartAsync(startAsync)
+            .build();
+  }
+
+  @Test
+  public void syntax_resumableBuffered_fluent() {
+    ApiFuture<ResumableWrite> startAsync =
+        ResumableMedia.gapic().write().resumableWrite(startResumableWrite, req);
+    BufferedWritableByteChannelSession<WriteObjectResponse> session =
+        ResumableMedia.gapic()
+            .write()
+            .byteChannel(write)
+            .setHasher(Hasher.noop())
+            .setByteStringStrategy(ByteStringStrategy.noCopy())
+            .resumable()
+            .buffered()
+            .setStartAsync(startAsync)
+            .build();
+  }
+
+  @Test
+  public void syntax_directUnbuffered_incremental() {
+    GapicWritableByteChannelSessionBuilder b1 =
+        ResumableMedia.gapic()
+            .write()
+            .byteChannel(write)
+            .setHasher(Hasher.noop())
+            .setByteStringStrategy(ByteStringStrategy.noCopy());
+    UnbufferedWritableByteChannelSession<WriteObjectResponse> session =
+        b1.direct().unbuffered().setRequest(req).build();
+  }
+
+  @Test
+  public void syntax_directBuffered_incremental() {
+    GapicWritableByteChannelSessionBuilder b1 =
+        ResumableMedia.gapic()
+            .write()
+            .byteChannel(write)
+            .setHasher(Hasher.noop())
+            .setByteStringStrategy(ByteStringStrategy.noCopy());
+    BufferedWritableByteChannelSession<WriteObjectResponse> session =
+        b1.direct().buffered().setRequest(req).build();
+  }
+
+  @Test
+  public void syntax_resumableUnbuffered_incremental() {
+    ApiFuture<ResumableWrite> startAsync =
+        ResumableMedia.gapic().write().resumableWrite(startResumableWrite, req);
+    GapicWritableByteChannelSessionBuilder b1 =
+        ResumableMedia.gapic()
+            .write()
+            .byteChannel(write)
+            .setHasher(Hasher.noop())
+            .setByteStringStrategy(ByteStringStrategy.noCopy());
+    UnbufferedWritableByteChannelSession<WriteObjectResponse> session =
+        b1.resumable().unbuffered().setStartAsync(startAsync).build();
+  }
+
+  @Test
+  public void syntax_resumableBuffered_incremental() {
+    ApiFuture<ResumableWrite> startAsync =
+        ResumableMedia.gapic().write().resumableWrite(startResumableWrite, req);
+    GapicWritableByteChannelSessionBuilder b1 =
+        ResumableMedia.gapic()
+            .write()
+            .byteChannel(write)
+            .setHasher(Hasher.noop())
+            .setByteStringStrategy(ByteStringStrategy.noCopy());
+    BufferedWritableByteChannelSession<WriteObjectResponse> session =
+        b1.resumable().buffered().setStartAsync(startAsync).build();
+  }
+}