Repository: googleapis/java-storage
Pull request #2143 (2023-07-27 18:41:56)

build(deps): bump certifi from 2023.5.7 to 2023.7.22 in /synthtool/gcp/templates/java_library/.kokoro (#1837)
build(deps): bump certifi

Bumps [certifi](https://github.com/certifi/python-certifi) from 2023.5.7 to 2023.7.22.
- [Commits](https://github.com/certifi/python-certifi/compare/2023.05.07...2023.07.22)

---
updated-dependencies:
- dependency-name: certifi
  dependency-type: indirect
...

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
Source-Link: https://github.com/googleapis/synthtool/commit/d85e1d678a829da6f2f5664392a6cd8e95ba8341
Post-Processor: gcr.io/cloud-devrel-public-resources/owlbot-java:latest@sha256:3a95f1b9b1102865ca551b76be51d2bdb850900c4db2f6d79269e7af81ac8f84


Patch:
diff --git a/.github/.OwlBot.lock.yaml b/.github/.OwlBot.lock.yaml
index d5500ef4..fa335912 100644
--- a/.github/.OwlBot.lock.yaml
+++ b/.github/.OwlBot.lock.yaml
@@ -13,5 +13,5 @@
 # limitations under the License.
 docker:
   image: gcr.io/cloud-devrel-public-resources/owlbot-java:latest
-  digest: sha256:46d2d262cd285c638656c8bde468011b723dc0c7ffd6a5ecc2650fe639c82e8f
-# created: 2023-07-24T14:21:17.707234503Z
+  digest: sha256:3a95f1b9b1102865ca551b76be51d2bdb850900c4db2f6d79269e7af81ac8f84
+# created: 2023-07-27T18:37:44.251188775Z
diff --git a/.kokoro/requirements.txt b/.kokoro/requirements.txt
index 32989051..a73256ab 100644
--- a/.kokoro/requirements.txt
+++ b/.kokoro/requirements.txt
@@ -12,9 +12,9 @@ cachetools==5.3.1 \
     --hash=sha256:95ef631eeaea14ba2e36f06437f36463aac3a096799e876ee55e5cdccb102590 \
     --hash=sha256:dce83f2d9b4e1f732a8cd44af8e8fab2dbe46201467fc98b3ef8f269092bf62b
     # via google-auth
-certifi==2023.5.7 \
-    --hash=sha256:0f0d56dc5a6ad56fd4ba36484d6cc34451e1c6548c61daad8c320169f91eddc7 \
-    --hash=sha256:c6c2e98f5c7869efca1f8916fed228dd91539f9f1b444c314c06eef02980c716
+certifi==2023.7.22 \
+    --hash=sha256:539cc1d13202e33ca466e88b2807e29f4c13049d6d87031a3c110744495cb082 \
+    --hash=sha256:92d6037539857d8206b8f6ae472e8b77db8058fec5937a1ef3f54304089edbb9
     # via requests
 cffi==1.15.1 \
     --hash=sha256:00a9ed42e88df81ffae7a8ab6d9356b371399b91dbdf0c3cb1e84c03a13aceb5 \
@@ -485,6 +485,5 @@ zipp==3.16.1 \
     # via importlib-metadata
 
 # WARNING: The following packages were not pinned, but pip requires them to be
-# pinned when the requirements file includes hashes and the requirement is not
-# satisfied by a package already installed. Consider using the --allow-unsafe flag.
+# pinned when the requirements file includes hashes. Consider using the --allow-unsafe flag.
 # setuptools
diff --git a/README.md b/README.md
index 1aa0a611..560db1b6 100644
--- a/README.md
+++ b/README.md
@@ -19,7 +19,7 @@ If you are using Maven with [BOM][libraries-bom], add this to your pom.xml file:
     <dependency>
       <groupId>com.google.cloud</groupId>
       <artifactId>libraries-bom</artifactId>
-      <version>26.19.0</version>
+      <version>26.21.0</version>
       <type>pom</type>
       <scope>import</scope>
     </dependency>
@@ -50,7 +50,7 @@ If you are using Maven without the BOM, add this to your dependencies:
 If you are using Gradle 5.x or later, add this to your dependencies:
 
 ```Groovy
-implementation platform('com.google.cloud:libraries-bom:26.19.0')
+implementation platform('com.google.cloud:libraries-bom:26.21.0')
 
 implementation 'com.google.cloud:google-cloud-storage'
 ```
diff --git a/google-cloud-storage/clirr-ignored-differences.xml b/google-cloud-storage/clirr-ignored-differences.xml
index 84d99070..c7189a7a 100644
--- a/google-cloud-storage/clirr-ignored-differences.xml
+++ b/google-cloud-storage/clirr-ignored-differences.xml
@@ -8,4 +8,11 @@
     <method>com.google.cloud.storage.BlobWriteSession blobWriteSession(com.google.cloud.storage.BlobInfo, com.google.cloud.storage.Storage$BlobWriteOption[])</method>
   </difference>
 
+  <!-- Not breaking, new method has a default implementation -->
+  <difference>
+    <differenceType>7012</differenceType>
+    <className>com/google/cloud/storage/UnbufferedWritableByteChannelSession$UnbufferedWritableByteChannel</className>
+    <method>* writeAndClose(*)</method>
+  </difference>
+
 </differences>
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/ApiaryUnbufferedWritableByteChannel.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/ApiaryUnbufferedWritableByteChannel.java
index 8201fbc6..da138572 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/ApiaryUnbufferedWritableByteChannel.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/ApiaryUnbufferedWritableByteChannel.java
@@ -57,6 +57,40 @@ final class ApiaryUnbufferedWritableByteChannel implements UnbufferedWritableByt
 
   @Override
   public long write(ByteBuffer[] srcs, int offset, int length) throws IOException {
+    return internalWrite(srcs, offset, length, false);
+  }
+
+  @Override
+  public long writeAndClose(ByteBuffer[] srcs, int offset, int length) throws IOException {
+    long write = internalWrite(srcs, offset, length, true);
+    close();
+    return write;
+  }
+
+  @Override
+  public boolean isOpen() {
+    return open;
+  }
+
+  @Override
+  public void close() throws IOException {
+    open = false;
+    if (!finished) {
+      try {
+        ResumableOperationResult<@Nullable StorageObject> operationResult =
+            session.put(RewindableContent.empty(), HttpContentRange.of(cumulativeByteCount));
+        long persistedSize = operationResult.getPersistedSize();
+        committedBytesCallback.accept(persistedSize);
+        result.set(operationResult.getObject());
+      } catch (Exception e) {
+        result.setException(e);
+        throw StorageException.coalesce(e);
+      }
+    }
+  }
+
+  private long internalWrite(ByteBuffer[] srcs, int offset, int length, boolean finalize)
+      throws ClosedChannelException {
     if (!open) {
       throw new ClosedChannelException();
     }
@@ -65,9 +99,13 @@ final class ApiaryUnbufferedWritableByteChannel implements UnbufferedWritableByt
     long newFinalByteOffset = cumulativeByteCount + available;
     final HttpContentRange header;
     ByteRangeSpec rangeSpec = ByteRangeSpec.explicit(cumulativeByteCount, newFinalByteOffset);
-    if (available % ByteSizeConstants._256KiB == 0) {
+    boolean quantumAligned = available % ByteSizeConstants._256KiB == 0;
+    if (quantumAligned && finalize) {
+      header = HttpContentRange.of(rangeSpec, newFinalByteOffset);
+      finished = true;
+    } else if (quantumAligned) {
       header = HttpContentRange.of(rangeSpec);
-    } else {
+    } else { // not quantum aligned, have to finalize
       header = HttpContentRange.of(rangeSpec, newFinalByteOffset);
       finished = true;
     }
@@ -87,26 +125,4 @@ final class ApiaryUnbufferedWritableByteChannel implements UnbufferedWritableByt
       throw StorageException.coalesce(e);
     }
   }
-
-  @Override
-  public boolean isOpen() {
-    return open;
-  }
-
-  @Override
-  public void close() throws IOException {
-    open = false;
-    if (!finished) {
-      try {
-        ResumableOperationResult<@Nullable StorageObject> operationResult =
-            session.put(RewindableContent.empty(), HttpContentRange.of(cumulativeByteCount));
-        long persistedSize = operationResult.getPersistedSize();
-        committedBytesCallback.accept(persistedSize);
-        result.set(operationResult.getObject());
-      } catch (Exception e) {
-        result.setException(e);
-        throw StorageException.coalesce(e);
-      }
-    }
-  }
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteSessionConfig.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteSessionConfig.java
index de8622c7..a67d5a20 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteSessionConfig.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteSessionConfig.java
@@ -23,6 +23,7 @@ import com.google.cloud.storage.UnifiedOpts.ObjectTargetOpt;
 import com.google.cloud.storage.UnifiedOpts.Opts;
 import com.google.storage.v2.WriteObjectResponse;
 import java.io.IOException;
+import java.io.Serializable;
 import java.time.Clock;
 
 /**
@@ -39,7 +40,7 @@ import java.time.Clock;
  */
 // When we have java modules, actually seal this to internal extension only
 @InternalApi
-public abstract class BlobWriteSessionConfig {
+public abstract class BlobWriteSessionConfig implements Serializable {
 
   @InternalApi
   BlobWriteSessionConfig() {}
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteSessionConfigs.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteSessionConfigs.java
index cc5e691e..efaf569a 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteSessionConfigs.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteSessionConfigs.java
@@ -19,6 +19,11 @@ package com.google.cloud.storage;
 import com.google.api.core.BetaApi;
 import com.google.cloud.storage.GrpcStorageOptions.GrpcStorageDefaults;
 import com.google.cloud.storage.Storage.BlobWriteOption;
+import com.google.common.collect.ImmutableList;
+import java.io.IOException;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.Collection;
 
 /**
  * Factory class to select and construct {@link BlobWriteSessionConfig}s.
@@ -46,4 +51,54 @@ public final class BlobWriteSessionConfigs {
   public static DefaultBlobWriteSessionConfig getDefault() {
     return new DefaultBlobWriteSessionConfig(ByteSizeConstants._16MiB);
   }
+
+  /**
+   * Create a new {@link BlobWriteSessionConfig} which will first buffer the content of the object
+   * to a temporary file under {@code java.io.tmpdir}.
+   *
+   * <p>Once the file on disk is closed, the entire file will then be uploaded to Google Cloud
+   * Storage.
+   *
+   * @see Storage#blobWriteSession(BlobInfo, BlobWriteOption...)
+   * @see GrpcStorageOptions.Builder#setBlobWriteSessionConfig(BlobWriteSessionConfig)
+   */
+  @BetaApi
+  public static BlobWriteSessionConfig bufferToTempDirThenUpload() throws IOException {
+    return bufferToDiskThenUpload(
+        Paths.get(System.getProperty("java.io.tmpdir"), "google-cloud-storage"));
+  }
+
+  /**
+   * Create a new {@link BlobWriteSessionConfig} which will first buffer the content of the object
+   * to a temporary file under the specified {@code path}.
+   *
+   * <p>Once the file on disk is closed, the entire file will then be uploaded to Google Cloud
+   * Storage.
+   *
+   * @see Storage#blobWriteSession(BlobInfo, BlobWriteOption...)
+   * @see GrpcStorageOptions.Builder#setBlobWriteSessionConfig(BlobWriteSessionConfig)
+   */
+  @BetaApi
+  public static BufferToDiskThenUpload bufferToDiskThenUpload(Path path) throws IOException {
+    return bufferToDiskThenUpload(ImmutableList.of(path));
+  }
+
+  /**
+   * Create a new {@link BlobWriteSessionConfig} which will first buffer the content of the object
+   * to a temporary file under one of the specified {@code paths}.
+   *
+   * <p>Once the file on disk is closed, the entire file will then be uploaded to Google Cloud
+   * Storage.
+   *
+   * <p>The specifics of how the work is spread across multiple paths is undefined and subject to
+   * change.
+   *
+   * @see Storage#blobWriteSession(BlobInfo, BlobWriteOption...)
+   * @see GrpcStorageOptions.Builder#setBlobWriteSessionConfig(BlobWriteSessionConfig)
+   */
+  @BetaApi
+  public static BufferToDiskThenUpload bufferToDiskThenUpload(Collection<Path> paths)
+      throws IOException {
+    return new BufferToDiskThenUpload(ImmutableList.copyOf(paths), false);
+  }
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/BufferToDiskThenUpload.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/BufferToDiskThenUpload.java
new file mode 100644
index 00000000..fb20747a
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/BufferToDiskThenUpload.java
@@ -0,0 +1,236 @@
+/*
+ * Copyright 2023 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.core.BetaApi;
+import com.google.api.core.InternalApi;
+import com.google.api.core.SettableApiFuture;
+import com.google.cloud.storage.Conversions.Decoder;
+import com.google.cloud.storage.Storage.BlobWriteOption;
+import com.google.cloud.storage.UnifiedOpts.ObjectTargetOpt;
+import com.google.cloud.storage.UnifiedOpts.Opts;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.ImmutableList;
+import com.google.common.util.concurrent.MoreExecutors;
+import com.google.storage.v2.WriteObjectResponse;
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.nio.ByteBuffer;
+import java.nio.channels.WritableByteChannel;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.time.Clock;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.stream.Collector;
+import javax.annotation.concurrent.Immutable;
+import org.checkerframework.checker.nullness.qual.MonotonicNonNull;
+
+/**
+ * There are scenarios in which disk space is more plentiful than memory space. This new {@link
+ * BlobWriteSessionConfig} allows augmenting an instance of storage to produce {@link
+ * BlobWriteSession}s which will buffer to disk rather than holding things in memory.
+ *
+ * <p>Once the file on disk is closed, the entire file will then be uploaded to GCS.
+ *
+ * @see Storage#blobWriteSession(BlobInfo, BlobWriteOption...)
+ * @see GrpcStorageOptions.Builder#setBlobWriteSessionConfig(BlobWriteSessionConfig)
+ * @see BlobWriteSessionConfigs#bufferToDiskThenUpload(Path)
+ * @see BlobWriteSessionConfigs#bufferToDiskThenUpload(Collection)
+ */
+@Immutable
+@BetaApi
+public final class BufferToDiskThenUpload extends BlobWriteSessionConfig {
+  private static final long serialVersionUID = 9059242302276891867L;
+
+  /**
+   * non-final because of {@link java.io.Serializable}, however this field is effectively final as
+   * it is immutable and there is not reference mutator method.
+   */
+  @MonotonicNonNull private transient ImmutableList<Path> paths;
+
+  private final boolean includeLoggingSink;
+
+  /** Used for {@link java.io.Serializable} */
+  @MonotonicNonNull private volatile ArrayList<String> absolutePaths;
+
+  @InternalApi
+  BufferToDiskThenUpload(ImmutableList<Path> paths, boolean includeLoggingSink) throws IOException {
+    this.paths = paths;
+    this.includeLoggingSink = includeLoggingSink;
+  }
+
+  @VisibleForTesting
+  @InternalApi
+  BufferToDiskThenUpload withIncludeLoggingSink() throws IOException {
+    return new BufferToDiskThenUpload(paths, true);
+  }
+
+  @InternalApi
+  @Override
+  WriterFactory createFactory(Clock clock) throws IOException {
+    Duration window = Duration.ofMinutes(10);
+    RecoveryFileManager recoveryFileManager =
+        RecoveryFileManager.of(paths, getRecoverVolumeSinkFactory(clock, window));
+    ThroughputSink gcs = ThroughputSink.windowed(ThroughputMovingWindow.of(window), clock);
+    gcs = includeLoggingSink ? ThroughputSink.tee(ThroughputSink.logged("gcs", clock), gcs) : gcs;
+    return new Factory(recoveryFileManager, clock, gcs);
+  }
+
+  private RecoveryFileManager.RecoverVolumeSinkFactory getRecoverVolumeSinkFactory(
+      Clock clock, Duration window) {
+    return path -> {
+      ThroughputSink windowed = ThroughputSink.windowed(ThroughputMovingWindow.of(window), clock);
+      if (includeLoggingSink) {
+        return ThroughputSink.tee(
+            ThroughputSink.logged(path.toAbsolutePath().toString(), clock), windowed);
+      } else {
+        return windowed;
+      }
+    };
+  }
+
+  private void writeObject(ObjectOutputStream out) throws IOException {
+    if (absolutePaths == null) {
+      synchronized (this) {
+        if (absolutePaths == null) {
+          absolutePaths =
+              paths.stream()
+                  .map(Path::toAbsolutePath)
+                  .map(Path::toString)
+                  .collect(
+                      Collector.of(
+                          ArrayList::new,
+                          ArrayList::add,
+                          (left, right) -> {
+                            left.addAll(right);
+                            return left;
+                          }));
+        }
+      }
+    }
+    out.defaultWriteObject();
+  }
+
+  private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
+    in.defaultReadObject();
+    this.paths = absolutePaths.stream().map(Paths::get).collect(ImmutableList.toImmutableList());
+  }
+
+  private static final class Factory implements WriterFactory {
+
+    private final RecoveryFileManager recoveryFileManager;
+    private final Clock clock;
+    private final ThroughputSink gcs;
+
+    private Factory(RecoveryFileManager recoveryFileManager, Clock clock, ThroughputSink gcs) {
+      this.recoveryFileManager = recoveryFileManager;
+      this.clock = clock;
+      this.gcs = gcs;
+    }
+
+    @InternalApi
+    @Override
+    public WritableByteChannelSession<?, BlobInfo> writeSession(
+        StorageInternal storage,
+        BlobInfo info,
+        Opts<ObjectTargetOpt> opts,
+        Decoder<WriteObjectResponse, BlobInfo> d) {
+      return new Factory.WriteToFileThenUpload(
+          storage, info, opts, recoveryFileManager.newRecoveryFile(info));
+    }
+
+    private final class WriteToFileThenUpload
+        implements WritableByteChannelSession<WritableByteChannel, BlobInfo> {
+
+      private final StorageInternal storage;
+      private final BlobInfo info;
+      private final Opts<ObjectTargetOpt> opts;
+      private final RecoveryFile rf;
+      private final SettableApiFuture<BlobInfo> result;
+
+      private WriteToFileThenUpload(
+          StorageInternal storage, BlobInfo info, Opts<ObjectTargetOpt> opts, RecoveryFile rf) {
+        this.info = info;
+        this.opts = opts;
+        this.rf = rf;
+        this.storage = storage;
+        this.result = SettableApiFuture.create();
+      }
+
+      @Override
+      public ApiFuture<WritableByteChannel> openAsync() {
+        try {
+          ApiFuture<WritableByteChannel> f = ApiFutures.immediateFuture(rf.writer());
+          return ApiFutures.transform(
+              f, Factory.WriteToFileThenUpload.Flusher::new, MoreExecutors.directExecutor());
+        } catch (IOException e) {
+          throw StorageException.coalesce(e);
+        }
+      }
+
+      @Override
+      public ApiFuture<BlobInfo> getResult() {
+        return result;
+      }
+
+      private final class Flusher implements WritableByteChannel {
+
+        private final WritableByteChannel delegate;
+
+        private Flusher(WritableByteChannel delegate) {
+          this.delegate = delegate;
+        }
+
+        @Override
+        public int write(ByteBuffer src) throws IOException {
+          return delegate.write(src);
+        }
+
+        @Override
+        public boolean isOpen() {
+          return delegate.isOpen();
+        }
+
+        @Override
+        public void close() throws IOException {
+          delegate.close();
+          try (RecoveryFile rf = Factory.WriteToFileThenUpload.this.rf) {
+            Path path = rf.getPath();
+            long size = Files.size(path);
+            ThroughputSink.computeThroughput(
+                clock,
+                gcs,
+                size,
+                () -> {
+                  BlobInfo blob = storage.internalCreateFrom(path, info, opts);
+                  result.set(blob);
+                });
+          } catch (StorageException | IOException e) {
+            result.setException(e);
+            throw e;
+          }
+        }
+      }
+    }
+  }
+}
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/DefaultBlobWriteSessionConfig.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/DefaultBlobWriteSessionConfig.java
index dfea0d41..9b4a3983 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/DefaultBlobWriteSessionConfig.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/DefaultBlobWriteSessionConfig.java
@@ -53,6 +53,7 @@ import javax.annotation.concurrent.Immutable;
 @Immutable
 @BetaApi
 public final class DefaultBlobWriteSessionConfig extends BlobWriteSessionConfig {
+  private static final long serialVersionUID = -6873740918589930633L;
 
   private final int chunkSize;
 
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/DefaultBufferedWritableByteChannel.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/DefaultBufferedWritableByteChannel.java
index 3514b7d6..f1b14d06 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/DefaultBufferedWritableByteChannel.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/DefaultBufferedWritableByteChannel.java
@@ -160,8 +160,17 @@ final class DefaultBufferedWritableByteChannel implements BufferedWritableByteCh
 
   @Override
   public void close() throws IOException {
-    try (UnbufferedWritableByteChannel ignored = channel) {
-      flush();
+    if (enqueuedBytes()) {
+      ByteBuffer buffer = handle.get();
+      Buffers.flip(buffer);
+      channel.writeAndClose(buffer);
+      if (buffer.hasRemaining()) {
+        buffer.compact();
+      } else {
+        Buffers.clear(buffer);
+      }
+    } else {
+      channel.close();
     }
   }
 
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/DefaultStorageRetryStrategy.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/DefaultStorageRetryStrategy.java
index fb3ac380..cf85b740 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/DefaultStorageRetryStrategy.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/DefaultStorageRetryStrategy.java
@@ -145,7 +145,7 @@ final class DefaultStorageRetryStrategy implements StorageRetryStrategy {
     public RetryResult beforeEval(Exception exception) {
       if (exception instanceof IllegalArgumentException) {
         IllegalArgumentException illegalArgumentException = (IllegalArgumentException) exception;
-        if (illegalArgumentException.getMessage().equals("no JSON input found")) {
+        if ("no JSON input found".equals(illegalArgumentException.getMessage())) {
           return RetryResult.RETRY;
         }
       }
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 951a52ed..ebe375b5 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
@@ -63,12 +63,60 @@ final class GapicUnbufferedWritableByteChannel<
   }
 
   @Override
-  public long write(ByteBuffer[] srcs, int srcsOffset, int srcLength) throws IOException {
+  public long write(ByteBuffer[] srcs, int srcsOffset, int srcsLength) throws IOException {
+    return internalWrite(srcs, srcsOffset, srcsLength, false);
+  }
+
+  @Override
+  public long writeAndClose(ByteBuffer[] srcs, int srcsOffset, int srcsLength) throws IOException {
+    long write = internalWrite(srcs, srcsOffset, srcsLength, true);
+    close();
+    return write;
+  }
+
+  @Override
+  public boolean isOpen() {
+    return open;
+  }
+
+  @Override
+  public void close() throws IOException {
+    if (!finished) {
+      long offset = writeCtx.getTotalSentBytes().get();
+      Crc32cLengthKnown crc32cValue = writeCtx.getCumulativeCrc32c().get();
+
+      WriteObjectRequest.Builder b =
+          writeCtx.newRequestBuilder().setFinishWrite(true).setWriteOffset(offset);
+      if (crc32cValue != null) {
+        b.setObjectChecksums(
+            ObjectChecksums.newBuilder().setCrc32C(crc32cValue.getValue()).build());
+      }
+      WriteObjectRequest message = b.build();
+      try {
+        flusher.close(message);
+        finished = true;
+      } catch (RuntimeException e) {
+        resultFuture.setException(e);
+        throw e;
+      }
+    } else {
+      flusher.close(null);
+    }
+    open = false;
+  }
+
+  @VisibleForTesting
+  WriteCtx<RequestFactoryT> getWriteCtx() {
+    return writeCtx;
+  }
+
+  private long internalWrite(ByteBuffer[] srcs, int srcsOffset, int srcsLength, boolean finalize)
+      throws ClosedChannelException {
     if (!open) {
       throw new ClosedChannelException();
     }
 
-    ChunkSegment[] data = chunkSegmenter.segmentBuffers(srcs, srcsOffset, srcLength);
+    ChunkSegment[] data = chunkSegmenter.segmentBuffers(srcs, srcsOffset, srcsLength);
 
     List<WriteObjectRequest> messages = new ArrayList<>();
 
@@ -91,7 +139,7 @@ final class GapicUnbufferedWritableByteChannel<
               .newRequestBuilder()
               .setWriteOffset(offset)
               .setChecksummedData(checksummedData.build());
-      if (!datum.isOnlyFullBlocks()) {
+      if (!datum.isOnlyFullBlocks() || finalize) {
         builder.setFinishWrite(true);
         if (cumulative != null) {
           builder.setObjectChecksums(
@@ -114,40 +162,4 @@ final class GapicUnbufferedWritableByteChannel<
 
     return bytesConsumed;
   }
-
-  @Override
-  public boolean isOpen() {
-    return open;
-  }
-
-  @Override
-  public void close() throws IOException {
-    if (!finished) {
-      long offset = writeCtx.getTotalSentBytes().get();
-      Crc32cLengthKnown crc32cValue = writeCtx.getCumulativeCrc32c().get();
-
-      WriteObjectRequest.Builder b =
-          writeCtx.newRequestBuilder().setFinishWrite(true).setWriteOffset(offset);
-      if (crc32cValue != null) {
-        b.setObjectChecksums(
-            ObjectChecksums.newBuilder().setCrc32C(crc32cValue.getValue()).build());
-      }
-      WriteObjectRequest message = b.build();
-      try {
-        flusher.close(message);
-        finished = true;
-      } catch (RuntimeException e) {
-        resultFuture.setException(e);
-        throw e;
-      }
-    } else {
-      flusher.close(null);
-    }
-    open = false;
-  }
-
-  @VisibleForTesting
-  WriteCtx<RequestFactoryT> getWriteCtx() {
-    return writeCtx;
-  }
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcResumableSession.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcResumableSession.java
index 8de44fb6..2c60c9fb 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcResumableSession.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcResumableSession.java
@@ -91,6 +91,7 @@ final class GrpcResumableSession {
             if (query.getObject() != null) {
               return query;
             } else {
+              handle.get().clear();
               content.rewindTo(query.getPersistedSize());
             }
           }
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 fdd67a7e..fad1f66e 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
@@ -152,7 +152,8 @@ import java.util.stream.StreamSupport;
 import org.checkerframework.checker.nullness.qual.Nullable;
 
 @BetaApi
-final class GrpcStorageImpl extends BaseService<StorageOptions> implements StorageInternal {
+final class GrpcStorageImpl extends BaseService<StorageOptions>
+    implements Storage, StorageInternal {
 
   private static final byte[] ZERO_BYTES = new byte[0];
   private static final Set<OpenOption> READ_OPS = ImmutableSet.of(StandardOpenOption.READ);
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageOptions.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageOptions.java
index 8bb3115c..a6aa3313 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageOptions.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageOptions.java
@@ -48,6 +48,7 @@ import com.google.cloud.storage.UnifiedOpts.Opts;
 import com.google.cloud.storage.UnifiedOpts.UserProject;
 import com.google.cloud.storage.spi.StorageRpcFactory;
 import com.google.common.base.MoreObjects;
+import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableSet;
 import com.google.storage.v2.ReadObjectRequest;
@@ -169,6 +170,7 @@ public final class GrpcStorageOptions extends StorageOptions
 
     Opts<UserProject> defaultOpts = Opts.empty();
     CredentialsProvider credentialsProvider;
+    Preconditions.checkState(credentials != null, "Unable to resolve credentials");
     if (credentials instanceof NoCredentials) {
       credentialsProvider = NoCredentialsProvider.create();
     } else {
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/RecoveryFile.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/RecoveryFile.java
new file mode 100644
index 00000000..d399ea93
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/RecoveryFile.java
@@ -0,0 +1,100 @@
+/*
+ * Copyright 2023 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.common.base.MoreObjects;
+import com.google.common.collect.ImmutableSet;
+import java.io.IOException;
+import java.nio.channels.SeekableByteChannel;
+import java.nio.channels.WritableByteChannel;
+import java.nio.file.Files;
+import java.nio.file.OpenOption;
+import java.nio.file.Path;
+import java.nio.file.StandardOpenOption;
+import java.util.Set;
+
+/**
+ * When uploading to GCS, there are times when memory buffers are not preferable. This class
+ * encapsulates the logic and lifecycle for a file written to local disk which can be used for
+ * upload recovery in the case an upload is interrupted.
+ */
+final class RecoveryFile implements AutoCloseable {
+  private static final Set<OpenOption> writeOps =
+      ImmutableSet.of(StandardOpenOption.CREATE, StandardOpenOption.WRITE);
+  private static final Set<OpenOption> readOps = ImmutableSet.of(StandardOpenOption.READ);
+
+  private final Path path;
+  private final ThroughputSink throughputSink;
+  private final Runnable onCloseCallback;
+
+  RecoveryFile(Path path, ThroughputSink throughputSink, Runnable onCloseCallback) {
+    this.path = path;
+    this.throughputSink = throughputSink;
+    this.onCloseCallback = onCloseCallback;
+  }
+
+  public Path getPath() {
+    return path;
+  }
+
+  public Path touch() throws IOException {
+    return Files.createFile(path);
+  }
+
+  public SeekableByteChannel reader() throws IOException {
+    return Files.newByteChannel(path, readOps);
+  }
+
+  public WritableByteChannel writer() throws IOException {
+    return throughputSink.decorate(Files.newByteChannel(path, writeOps));
+  }
+
+  @Override
+  public void close() throws IOException {
+    Files.delete(path);
+    onCloseCallback.run();
+  }
+
+  @Override
+  public String toString() {
+    return MoreObjects.toStringHelper(this)
+        .add("path", path)
+        .add("throughputSink", throughputSink)
+        .add("onCloseCallback", onCloseCallback)
+        .toString();
+  }
+
+  Unsafe unsafe() {
+    return new Unsafe();
+  }
+
+  final class Unsafe {
+    public Path touch() throws UnsafeIOException {
+      try {
+        return RecoveryFile.this.touch();
+      } catch (IOException e) {
+        throw new UnsafeIOException(e);
+      }
+    }
+  }
+
+  static final class UnsafeIOException extends RuntimeException {
+    private UnsafeIOException(IOException cause) {
+      super(cause);
+    }
+  }
+}
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/RecoveryFileManager.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/RecoveryFileManager.java
new file mode 100644
index 00000000..25303b5f
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/RecoveryFileManager.java
@@ -0,0 +1,106 @@
+/*
+ * Copyright 2023 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 com.google.common.base.Preconditions.checkArgument;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.primitives.Ints;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Base64;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+final class RecoveryFileManager {
+
+  private final ImmutableList<RecoveryVolume> volumes;
+  /** Keep track of active info and file */
+  private final Map<BlobInfo, RecoveryFile> files;
+
+  /**
+   * Round-robin assign recovery files to the configured volumes. Use this index to keep track of
+   * which volume to assign to next.
+   */
+  private int nextVolumeIndex;
+
+  private RecoveryFileManager(List<RecoveryVolume> volumes) {
+    this.volumes = ImmutableList.copyOf(volumes);
+    this.files = Collections.synchronizedMap(new HashMap<>());
+    this.nextVolumeIndex = 0;
+  }
+
+  public RecoveryFile newRecoveryFile(BlobInfo info) {
+    int i = getNextVolumeIndex();
+    RecoveryVolume v = volumes.get(i);
+    int hashCode = info.hashCode();
+    String fileName = Base64.getUrlEncoder().encodeToString(Ints.toByteArray(hashCode));
+    Path path = v.basePath.resolve(fileName);
+    RecoveryFile recoveryFile = new RecoveryFile(path, v.sink, () -> files.remove(info));
+    files.put(info, recoveryFile);
+    return recoveryFile;
+  }
+
+  private synchronized int getNextVolumeIndex() {
+    return nextVolumeIndex = (nextVolumeIndex + 1) % volumes.size();
+  }
+
+  static RecoveryFileManager of(List<Path> volumes) throws IOException {
+    return of(volumes, p -> ThroughputSink.nullSink());
+  }
+
+  static RecoveryFileManager of(List<Path> volumes, RecoverVolumeSinkFactory factory)
+      throws IOException {
+    checkArgument(!volumes.isEmpty(), "At least one volume must be specified");
+    checkArgument(
+        volumes.stream().allMatch(p -> !Files.exists(p) || Files.isDirectory(p)),
+        "All provided volumes must either:\n1. Not yet exists\n2. Be directories");
+
+    for (Path v : volumes) {
+      if (!Files.exists(v)) {
+        Files.createDirectories(v);
+      }
+    }
+    ImmutableList<RecoveryVolume> recoveryVolumes =
+        volumes.stream()
+            .map(p -> RecoveryVolume.of(p, factory.apply(p)))
+            .collect(ImmutableList.toImmutableList());
+    return new RecoveryFileManager(recoveryVolumes);
+  }
+
+  @FunctionalInterface
+  interface RecoverVolumeSinkFactory {
+    ThroughputSink apply(Path p);
+  }
+
+  static final class RecoveryVolume {
+    private final Path basePath;
+    private final ThroughputSink sink;
+
+    private RecoveryVolume(Path basePath, ThroughputSink sink) {
+      this.basePath = basePath;
+      this.sink = sink;
+    }
+
+    public static RecoveryVolume of(Path basePath, ThroughputSink sink) {
+      return new RecoveryVolume(basePath, sink);
+    }
+  }
+}
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 c3ff8013..fb14d2e8 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
@@ -184,6 +184,22 @@ final class StorageByteChannels {
       return delegate.write(srcs, offset, length);
     }
 
+    @Override
+    public synchronized int writeAndClose(ByteBuffer src) throws IOException {
+      return delegate.writeAndClose(src);
+    }
+
+    @Override
+    public synchronized long writeAndClose(ByteBuffer[] srcs) throws IOException {
+      return delegate.writeAndClose(srcs);
+    }
+
+    @Override
+    public synchronized long writeAndClose(ByteBuffer[] srcs, int offset, int length)
+        throws IOException {
+      return delegate.writeAndClose(srcs, offset, length);
+    }
+
     @Override
     public boolean isOpen() {
       return delegate.isOpen();
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageInternal.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageInternal.java
index deb8a050..53fdc4e9 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageInternal.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageInternal.java
@@ -21,9 +21,10 @@ import com.google.cloud.storage.UnifiedOpts.Opts;
 import java.io.IOException;
 import java.nio.file.Path;
 
-interface StorageInternal extends Storage {
+interface StorageInternal {
 
-  Blob internalCreateFrom(Path path, BlobInfo info, Opts<ObjectTargetOpt> opts) throws IOException;
+  BlobInfo internalCreateFrom(Path path, BlobInfo info, Opts<ObjectTargetOpt> opts)
+      throws IOException;
 
   StorageWriteChannel internalWriter(BlobInfo info, Opts<ObjectTargetOpt> opts);
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/Throughput.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/Throughput.java
new file mode 100644
index 00000000..f0f54140
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/Throughput.java
@@ -0,0 +1,94 @@
+/*
+ * Copyright 2023 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.common.base.MoreObjects;
+import java.time.Duration;
+import java.util.Objects;
+
+/**
+ * Convenience class to encapsulate the concept of a throughput value.
+ *
+ * <p>Given a number of bytes and a duration compute the number of bytes per second.
+ */
+final class Throughput {
+
+  private static final double NANOS_PER_SECOND = 1_000_000_000d;
+  private final long numBytes;
+  private final Duration duration;
+
+  // TODO: is there a efficient way we can limit precision without having to use BigDecimal?
+  //   Realistically, we don't need precision smaller than 1 byte per microsecond, leading to
+  //   6 digits past the decimal of needed precision.
+  private final double bytesPerSecond;
+
+  private Throughput(long numBytes, Duration duration) {
+    this.numBytes = numBytes;
+    this.duration = duration;
+    this.bytesPerSecond = numBytes / (duration.toNanos() / NANOS_PER_SECOND);
+  }
+
+  public long getNumBytes() {
+    return numBytes;
+  }
+
+  public Duration getDuration() {
+    return duration;
+  }
+
+  public double toBps() {
+    return bytesPerSecond;
+  }
+
+  public Throughput plus(Throughput other) {
+    return new Throughput(this.numBytes + other.numBytes, this.duration.plus(other.duration));
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (!(o instanceof Throughput)) {
+      return false;
+    }
+    Throughput that = (Throughput) o;
+    return Double.compare(that.bytesPerSecond, bytesPerSecond) == 0;
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(bytesPerSecond);
+  }
+
+  @Override
+  public String toString() {
+    return MoreObjects.toStringHelper(this).add("bytesPerSecond", bytesPerSecond).toString();
+  }
+
+  public static Throughput zero() {
+    return new Throughput(0, Duration.ZERO);
+  }
+
+  public static Throughput of(long numBytes, Duration duration) {
+    return new Throughput(numBytes, duration);
+  }
+
+  public static Throughput bytesPerSecond(long numBytes) {
+    return new Throughput(numBytes, Duration.ofSeconds(1));
+  }
+}
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/ThroughputMovingWindow.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/ThroughputMovingWindow.java
new file mode 100644
index 00000000..6afae991
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/ThroughputMovingWindow.java
@@ -0,0 +1,98 @@
+/*
+ * Copyright 2023 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.common.base.MoreObjects;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.Comparator;
+import java.util.PriorityQueue;
+
+/**
+ * A simple moving window implementation which will keep a {@code window}s worth of Throughput
+ * values and allow querying for the aggregate avg over that time window.
+ */
+final class ThroughputMovingWindow {
+
+  private final Duration window;
+
+  private final PriorityQueue<Entry> values;
+
+  private ThroughputMovingWindow(Duration window) {
+    this.window = window;
+    this.values = new PriorityQueue<>(Entry.COMP);
+  }
+
+  void add(Instant now, Throughput value) {
+    removeExpiredEntries(now);
+    values.add(new Entry(now, value));
+  }
+
+  Throughput avg(Instant now) {
+    removeExpiredEntries(now);
+    return values.stream()
+        .map(Entry::getValue)
+        .reduce(
+            Throughput.zero(),
+            (tp1, tp2) -> Throughput.of(tp1.getNumBytes() + tp2.getNumBytes(), window));
+  }
+
+  private void removeExpiredEntries(Instant now) {
+    Instant newMin = now.minus(window);
+    values.removeIf(e -> lteq(e.getAt(), newMin));
+  }
+
+  @Override
+  public String toString() {
+    return MoreObjects.toStringHelper(this)
+        .add("window", window)
+        .add("values.size()", values.size())
+        .toString();
+  }
+
+  static ThroughputMovingWindow of(Duration window) {
+    return new ThroughputMovingWindow(window);
+  }
+
+  private static boolean lteq(Instant a, Instant b) {
+    return a.equals(b) || a.isBefore(b);
+  }
+
+  private static final class Entry {
+    private static final Comparator<Entry> COMP = Comparator.comparing(e -> e.at);
+    private final Instant at;
+    private final Throughput value;
+
+    private Entry(Instant at, Throughput value) {
+      this.at = at;
+      this.value = value;
+    }
+
+    public Instant getAt() {
+      return at;
+    }
+
+    public Throughput getValue() {
+      return value;
+    }
+
+    @Override
+    public String toString() {
+      return MoreObjects.toStringHelper(this).add("at", at).add("value", value).toString();
+    }
+  }
+}
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/ThroughputSink.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/ThroughputSink.java
new file mode 100644
index 00000000..776629cf
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/ThroughputSink.java
@@ -0,0 +1,293 @@
+/*
+ * Copyright 2023 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.common.base.MoreObjects;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.WritableByteChannel;
+import java.time.Clock;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.Objects;
+import java.util.logging.Logger;
+
+/**
+ * Interface to mark a location in which throughput of byte movements can be recorded, and which can
+ * provide a decorated underlying channel.
+ */
+interface ThroughputSink {
+
+  void recordThroughput(Record r);
+
+  WritableByteChannel decorate(WritableByteChannel wbc);
+
+  static void computeThroughput(Clock clock, ThroughputSink sink, long numBytes, IO io)
+      throws IOException {
+    boolean exception = false;
+    Instant begin = clock.instant();
+    try {
+      io.apply();
+    } catch (IOException e) {
+      exception = true;
+      throw e;
+    } finally {
+      Instant end = clock.instant();
+      Record record = Record.of(numBytes, begin, end, exception);
+      sink.recordThroughput(record);
+    }
+  }
+
+  @FunctionalInterface
+  interface IO {
+    void apply() throws IOException;
+  }
+
+  static ThroughputSink logged(String prefix, Clock clock) {
+    return new LoggedThroughputSink(prefix, clock);
+  }
+
+  static ThroughputSink windowed(ThroughputMovingWindow w, Clock clock) {
+    return new ThroughputMovingWindowThroughputSink(w, clock);
+  }
+
+  static ThroughputSink tee(ThroughputSink a, ThroughputSink b) {
+    return new TeeThroughputSink(a, b);
+  }
+
+  static ThroughputSink nullSink() {
+    return NullThroughputSink.INSTANCE;
+  }
+
+  final class Record {
+    private final long numBytes;
+    private final Instant begin;
+    private final Instant end;
+    private final boolean exception;
+
+    private Record(long numBytes, Instant begin, Instant end, boolean exception) {
+      this.numBytes = numBytes;
+      this.begin = begin;
+      this.end = end;
+      this.exception = exception;
+    }
+
+    public long getNumBytes() {
+      return numBytes;
+    }
+
+    public Instant getBegin() {
+      return begin;
+    }
+
+    public Instant getEnd() {
+      return end;
+    }
+
+    public Duration getDuration() {
+      return Duration.between(begin, end);
+    }
+
+    public boolean isException() {
+      return exception;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+      if (this == o) {
+        return true;
+      }
+      if (!(o instanceof Record)) {
+        return false;
+      }
+      Record record = (Record) o;
+      return numBytes == record.numBytes
+          && exception == record.exception
+          && Objects.equals(begin, record.begin)
+          && Objects.equals(end, record.end);
+    }
+
+    @Override
+    public int hashCode() {
+      return Objects.hash(numBytes, begin, end, exception);
+    }
+
+    @Override
+    public String toString() {
+      return MoreObjects.toStringHelper(this)
+          .add("numBytes", numBytes)
+          .add("begin", begin)
+          .add("end", end)
+          .add("exception", exception)
+          .toString();
+    }
+
+    public static Record of(long numBytes, Instant begin, Instant end, boolean exception) {
+      return new Record(numBytes, begin, end, exception);
+    }
+  }
+
+  final class LoggedThroughputSink implements ThroughputSink {
+    private static final Logger LOGGER = Logger.getLogger(ThroughputSink.class.getName());
+
+    private final String prefix;
+    private final Clock clock;
+
+    private LoggedThroughputSink(String prefix, Clock clock) {
+      this.prefix = prefix;
+      this.clock = clock;
+    }
+
+    private static final double MiB = 1d / (1024 * 1024);
+
+    @Override
+    public void recordThroughput(Record r) {
+      LOGGER.info(
+          () ->
+              String.format(
+                  "{%s} (%01.03f MiB/s) %s",
+                  prefix,
+                  ((r.numBytes * MiB)
+                      / (Duration.between(r.getBegin(), r.getEnd()).toMillis() / 1000d)),
+                  r));
+    }
+
+    @Override
+    public WritableByteChannel decorate(WritableByteChannel wbc) {
+      return new ThroughputRecordingWritableByteChannel(wbc, this, clock);
+    }
+
+    @Override
+    public String toString() {
+      return MoreObjects.toStringHelper(this).add("prefix", prefix).add("clock", clock).toString();
+    }
+  }
+
+  final class ThroughputRecordingWritableByteChannel implements WritableByteChannel {
+    private final WritableByteChannel delegate;
+    private final ThroughputSink sink;
+    private final Clock clock;
+
+    private ThroughputRecordingWritableByteChannel(
+        WritableByteChannel delegate, ThroughputSink sink, Clock clock) {
+      this.delegate = delegate;
+      this.sink = sink;
+      this.clock = clock;
+    }
+
+    @Override
+    public int write(ByteBuffer src) throws IOException {
+      boolean exception = false;
+      int remaining = src.remaining();
+      Instant begin = clock.instant();
+      try {
+        return delegate.write(src);
+      } catch (IOException e) {
+        exception = true;
+        throw e;
+      } finally {
+        Instant end = clock.instant();
+        Record record = Record.of(remaining - src.remaining(), begin, end, exception);
+        sink.recordThroughput(record);
+      }
+    }
+
+    @Override
+    public boolean isOpen() {
+      return delegate.isOpen();
+    }
+
+    @Override
+    public void close() throws IOException {
+      delegate.close();
+    }
+
+    @Override
+    public String toString() {
+      return MoreObjects.toStringHelper(this)
+          .add("delegate", delegate)
+          .add("sink", sink)
+          .add("clock", clock)
+          .toString();
+    }
+  }
+
+  final class TeeThroughputSink implements ThroughputSink {
+    private final ThroughputSink a;
+    private final ThroughputSink b;
+
+    private TeeThroughputSink(ThroughputSink a, ThroughputSink b) {
+      this.a = a;
+      this.b = b;
+    }
+
+    @Override
+    public void recordThroughput(Record r) {
+      a.recordThroughput(r);
+      b.recordThroughput(r);
+    }
+
+    @Override
+    public WritableByteChannel decorate(WritableByteChannel wbc) {
+      return b.decorate(a.decorate(wbc));
+    }
+
+    @Override
+    public String toString() {
+      return MoreObjects.toStringHelper(this).add("a", a).add("b", b).toString();
+    }
+  }
+
+  final class ThroughputMovingWindowThroughputSink implements ThroughputSink {
+    private final ThroughputMovingWindow w;
+    private final Clock clock;
+
+    private ThroughputMovingWindowThroughputSink(ThroughputMovingWindow w, Clock clock) {
+      this.w = w;
+      this.clock = clock;
+    }
+
+    @Override
+    public synchronized void recordThroughput(Record r) {
+      w.add(r.end, Throughput.of(r.getNumBytes(), r.getDuration()));
+    }
+
+    @Override
+    public WritableByteChannel decorate(WritableByteChannel wbc) {
+      return new ThroughputRecordingWritableByteChannel(wbc, this, clock);
+    }
+
+    @Override
+    public String toString() {
+      return MoreObjects.toStringHelper(this).add("w", w).add("clock", clock).toString();
+    }
+  }
+
+  final class NullThroughputSink implements ThroughputSink {
+    private static final NullThroughputSink INSTANCE = new NullThroughputSink();
+
+    private NullThroughputSink() {}
+
+    @Override
+    public void recordThroughput(Record r) {}
+
+    @Override
+    public WritableByteChannel decorate(WritableByteChannel wbc) {
+      return wbc;
+    }
+  }
+}
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/UnbufferedWritableByteChannelSession.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/UnbufferedWritableByteChannelSession.java
index 8affde6b..d7a5fcef 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/UnbufferedWritableByteChannelSession.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/UnbufferedWritableByteChannelSession.java
@@ -28,12 +28,26 @@ interface UnbufferedWritableByteChannelSession<ResultT>
   interface UnbufferedWritableByteChannel extends WritableByteChannel, GatheringByteChannel {
     @Override
     default int write(ByteBuffer src) throws IOException {
-      return Math.toIntExact(write(new ByteBuffer[] {src}));
+      return Math.toIntExact(write(new ByteBuffer[] {src}, 0, 1));
     }
 
     @Override
     default long write(ByteBuffer[] srcs) throws IOException {
       return write(srcs, 0, srcs.length);
     }
+
+    default int writeAndClose(ByteBuffer src) throws IOException {
+      return Math.toIntExact(writeAndClose(new ByteBuffer[] {src}, 0, 1));
+    }
+
+    default long writeAndClose(ByteBuffer[] srcs) throws IOException {
+      return writeAndClose(srcs, 0, srcs.length);
+    }
+
+    default long writeAndClose(ByteBuffer[] srcs, int offset, int length) throws IOException {
+      long write = write(srcs, offset, length);
+      close();
+      return write;
+    }
   }
 }
diff --git a/pom.xml b/pom.xml
index 787a218a..ad7e2756 100644
--- a/pom.xml
+++ b/pom.xml
@@ -59,13 +59,6 @@
 
   <dependencyManagement>
     <dependencies>
-      <dependency>
-        <groupId>io.grpc</groupId>
-        <artifactId>grpc-bom</artifactId>
-        <version>1.56.1</version>
-        <type>pom</type>
-        <scope>import</scope>
-      </dependency>
       <dependency>
         <groupId>com.google.cloud</groupId>
         <artifactId>google-cloud-shared-dependencies</artifactId>
diff --git a/samples/native-image-sample/pom.xml b/samples/native-image-sample/pom.xml
index d2ab7237..654259dc 100644
--- a/samples/native-image-sample/pom.xml
+++ b/samples/native-image-sample/pom.xml
@@ -29,7 +29,7 @@
       <dependency>
         <groupId>com.google.cloud</groupId>
         <artifactId>libraries-bom</artifactId>
-        <version>26.19.0</version>
+        <version>26.21.0</version>
         <type>pom</type>
         <scope>import</scope>
       </dependency>
diff --git a/samples/snippets/pom.xml b/samples/snippets/pom.xml
index 22fdfdb1..e12e2124 100644
--- a/samples/snippets/pom.xml
+++ b/samples/snippets/pom.xml
@@ -31,7 +31,7 @@
       <dependency>
         <groupId>com.google.cloud</groupId>
         <artifactId>libraries-bom</artifactId>
-        <version>26.19.0</version>
+        <version>26.21.0</version>
         <type>pom</type>
         <scope>import</scope>
       </dependency>


Test patch:
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/BufferToDiskThenUploadTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/BufferToDiskThenUploadTest.java
new file mode 100644
index 00000000..522be99b
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/BufferToDiskThenUploadTest.java
@@ -0,0 +1,87 @@
+/*
+ * Copyright 2023 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 com.google.cloud.storage.TestUtils.xxd;
+import static com.google.common.truth.Truth.assertThat;
+
+import com.google.cloud.storage.BlobWriteSessionConfig.WriterFactory;
+import com.google.cloud.storage.Conversions.Decoder;
+import com.google.cloud.storage.UnifiedOpts.ObjectTargetOpt;
+import com.google.cloud.storage.UnifiedOpts.Opts;
+import com.google.storage.v2.WriteObjectResponse;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.WritableByteChannel;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.concurrent.atomic.AtomicReference;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.rules.TestName;
+
+public final class BufferToDiskThenUploadTest {
+
+  private static final Decoder<WriteObjectResponse, BlobInfo>
+      WRITE_OBJECT_RESPONSE_BLOB_INFO_DECODER =
+          Conversions.grpc().blobInfo().compose(WriteObjectResponse::getResource);
+  @Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder();
+  @Rule public final TestName testName = new TestName();
+
+  @Test
+  public void happyPath() throws IOException {
+    Path tempDir = temporaryFolder.newFolder(testName.getMethodName()).toPath();
+
+    BufferToDiskThenUpload btdtu =
+        BlobWriteSessionConfigs.bufferToDiskThenUpload(tempDir).withIncludeLoggingSink();
+    TestClock clock = TestClock.tickBy(Instant.EPOCH, Duration.ofSeconds(1));
+    WriterFactory factory = btdtu.createFactory(clock);
+
+    BlobInfo blobInfo = BlobInfo.newBuilder("bucket", "object").build();
+    AtomicReference<byte[]> actualBytes = new AtomicReference<>(null);
+    WritableByteChannelSession<?, BlobInfo> writeSession =
+        factory.writeSession(
+            new StorageInternal() {
+              @Override
+              public BlobInfo internalCreateFrom(
+                  Path path, BlobInfo info, Opts<ObjectTargetOpt> opts) throws IOException {
+                byte[] actual = Files.readAllBytes(path);
+                actualBytes.compareAndSet(null, actual);
+                return info;
+              }
+
+              @Override
+              public StorageWriteChannel internalWriter(BlobInfo info, Opts<ObjectTargetOpt> opts) {
+                return null;
+              }
+            },
+            blobInfo,
+            Opts.empty(),
+            WRITE_OBJECT_RESPONSE_BLOB_INFO_DECODER);
+
+    byte[] bytes = DataGenerator.base64Characters().genBytes(128);
+    try (WritableByteChannel open = writeSession.open()) {
+      open.write(ByteBuffer.wrap(bytes));
+    }
+    String xxdActual = xxd(actualBytes.get());
+    String xxdExpected = xxd(bytes);
+    assertThat(xxdActual).isEqualTo(xxdExpected);
+  }
+}
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/DefaultBufferedWritableByteChannelTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/DefaultBufferedWritableByteChannelTest.java
index 23d5b984..2653d27f 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/DefaultBufferedWritableByteChannelTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/DefaultBufferedWritableByteChannelTest.java
@@ -17,9 +17,11 @@
 package com.google.cloud.storage;
 
 import static com.google.cloud.storage.ChunkSegmenterTest.TestData.fmt;
+import static com.google.cloud.storage.TestUtils.xxd;
 import static com.google.common.truth.Truth.assertThat;
 import static com.google.common.truth.Truth.assertWithMessage;
 import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.fail;
 
 import com.google.cloud.storage.BufferedWritableByteChannelSession.BufferedWritableByteChannel;
 import com.google.cloud.storage.UnbufferedWritableByteChannelSession.UnbufferedWritableByteChannel;
@@ -36,6 +38,7 @@ import java.util.Arrays;
 import java.util.Deque;
 import java.util.List;
 import java.util.Objects;
+import java.util.concurrent.atomic.AtomicBoolean;
 import net.jqwik.api.Arbitraries;
 import net.jqwik.api.Arbitrary;
 import net.jqwik.api.Combinators;
@@ -343,6 +346,60 @@ public final class DefaultBufferedWritableByteChannelTest {
     assertThat(actual).isEqualTo(expected);
   }
 
+  @Example
+  @SuppressWarnings("JUnit5AssertionsConverter")
+  void callingCloseWithBufferedDataShouldCallWriteAndClose() throws IOException {
+    ByteArrayOutputStream baos = new ByteArrayOutputStream();
+
+    AtomicBoolean closed = new AtomicBoolean(false);
+    UnbufferedWritableByteChannel delegate =
+        new UnbufferedWritableByteChannel() {
+          @Override
+          public long write(ByteBuffer[] srcs, int offset, int length) throws IOException {
+            fail("unexpected write(ByteBuffer[], int, int) call");
+            return 0;
+          }
+
+          @Override
+          public long writeAndClose(ByteBuffer[] srcs, int offset, int length) throws IOException {
+            long total = 0;
+            try (WritableByteChannel out = Channels.newChannel(baos)) {
+              for (ByteBuffer src : srcs) {
+                total += out.write(src);
+              }
+            }
+            closed.compareAndSet(false, true);
+            return total;
+          }
+
+          @Override
+          public boolean isOpen() {
+            return !closed.get();
+          }
+
+          @Override
+          public void close() throws IOException {
+            fail("unexpected close() call");
+          }
+        };
+    DefaultBufferedWritableByteChannel test =
+        new DefaultBufferedWritableByteChannel(BufferHandle.allocate(20), delegate);
+
+    byte[] bytes = DataGenerator.base64Characters().genBytes(10);
+    String expected = xxd(bytes);
+
+    int write = test.write(ByteBuffer.wrap(bytes));
+    assertThat(write).isEqualTo(10);
+
+    assertThat(closed.get()).isFalse();
+
+    test.close();
+
+    String actual = xxd(baos.toByteArray());
+    assertThat(actual).isEqualTo(expected);
+    assertThat(closed.get()).isTrue();
+  }
+
   @Property
   void bufferAllocationShouldOnlyHappenWhenNeeded(@ForAll("BufferSizes") WriteOps writeOps)
       throws IOException {
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/ITGapicUnbufferedWritableByteChannelTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITGapicUnbufferedWritableByteChannelTest.java
index 08d228c2..3b2bf1bf 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/ITGapicUnbufferedWritableByteChannelTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITGapicUnbufferedWritableByteChannelTest.java
@@ -17,6 +17,7 @@
 package com.google.cloud.storage;
 
 import static com.google.cloud.storage.TestUtils.apiException;
+import static com.google.cloud.storage.TestUtils.getChecksummedData;
 import static com.google.common.truth.Truth.assertThat;
 
 import com.google.api.core.SettableApiFuture;
@@ -25,6 +26,7 @@ import com.google.api.gax.rpc.DataLossException;
 import com.google.api.gax.rpc.PermissionDeniedException;
 import com.google.cloud.storage.Retrying.RetryingDependencies;
 import com.google.cloud.storage.WriteCtx.WriteObjectRequestBuilderFactory;
+import com.google.cloud.storage.WriteFlushStrategy.Flusher;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
@@ -48,10 +50,13 @@ import java.util.Set;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.BiConsumer;
 import java.util.logging.Logger;
 import java.util.stream.Collector;
 import java.util.stream.Collectors;
+import org.checkerframework.checker.nullness.qual.NonNull;
+import org.checkerframework.checker.nullness.qual.Nullable;
 import org.junit.Test;
 
 public final class ITGapicUnbufferedWritableByteChannelTest {
@@ -307,6 +312,52 @@ public final class ITGapicUnbufferedWritableByteChannelTest {
     assertThat(writeCtx.getConfirmedBytes().get()).isEqualTo(40);
   }
 
+  @Test
+  public void resumableUpload_finalizeWhenWriteAndCloseCalledEvenWhenQuantumAligned()
+      throws IOException, InterruptedException, ExecutionException {
+    int quantum = 10;
+    ChunkSegmenter segmenter =
+        new ChunkSegmenter(Hasher.noop(), ByteStringStrategy.copy(), 50, quantum);
+    SettableApiFuture<WriteObjectResponse> result = SettableApiFuture.create();
+
+    AtomicReference<List<WriteObjectRequest>> actualFlush = new AtomicReference<>();
+    WriteObjectRequest closeRequestSentinel =
+        WriteObjectRequest.newBuilder().setUploadId("sentinel").build();
+    AtomicReference<WriteObjectRequest> actualClose = new AtomicReference<>(closeRequestSentinel);
+    GapicUnbufferedWritableByteChannel<?> c =
+        new GapicUnbufferedWritableByteChannel<>(
+            result,
+            segmenter,
+            reqFactory,
+            (bucketName, committedTotalBytesCallback, onSuccessCallback) ->
+                new Flusher() {
+                  @Override
+                  public void flush(@NonNull List<WriteObjectRequest> segments) {
+                    actualFlush.compareAndSet(null, segments);
+                  }
+
+                  @Override
+                  public void close(@Nullable WriteObjectRequest req) {
+                    actualClose.compareAndSet(closeRequestSentinel, req);
+                  }
+                });
+
+    byte[] bytes = DataGenerator.base64Characters().genBytes(quantum);
+
+    long written = c.writeAndClose(ByteBuffer.wrap(bytes));
+    WriteObjectRequest expectedRequest =
+        WriteObjectRequest.newBuilder()
+            .setUploadId(uploadId)
+            .setChecksummedData(getChecksummedData(ByteString.copyFrom(bytes), Hasher.noop()))
+            .setFinishWrite(true)
+            .build();
+
+    assertThat(written).isEqualTo(10);
+    assertThat(actualFlush.get()).isEqualTo(ImmutableList.of(expectedRequest));
+    // calling close is okay, as long as the provided request is null
+    assertThat(actualClose.get()).isAnyOf(closeRequestSentinel, null);
+  }
+
   static class DirectWriteService extends StorageImplBase {
     private static final Logger LOGGER = Logger.getLogger(DirectWriteService.class.getName());
     private final BiConsumer<StreamObserver<WriteObjectResponse>, List<WriteObjectRequest>> c;
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/RecoveryFileManagerTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/RecoveryFileManagerTest.java
new file mode 100644
index 00000000..abf68893
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/RecoveryFileManagerTest.java
@@ -0,0 +1,146 @@
+/*
+ * Copyright 2023 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 com.google.common.truth.Truth.assertThat;
+import static org.junit.Assert.assertThrows;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.io.ByteStreams;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.Channels;
+import java.nio.channels.SeekableByteChannel;
+import java.nio.channels.WritableByteChannel;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.Objects;
+import java.util.stream.Stream;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.rules.TestName;
+
+public final class RecoveryFileManagerTest {
+  private static final int _128KiB = 128 * 1024;
+
+  @Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+  @Rule public final TestName testName = new TestName();
+
+  private final TestClock clock = TestClock.tickBy(Instant.EPOCH, Duration.ofSeconds(1));
+
+  @Test
+  public void happyPath() throws IOException {
+    Path tempDir = temporaryFolder.newFolder(testName.getMethodName()).toPath();
+    RecoveryFileManager rfm =
+        RecoveryFileManager.of(
+            ImmutableList.of(tempDir),
+            path -> ThroughputSink.logged(path.toAbsolutePath().toString(), clock));
+
+    BlobInfo info = BlobInfo.newBuilder("bucket", "object").build();
+    try (RecoveryFile recoveryFile = rfm.newRecoveryFile(info)) {
+
+      byte[] bytes = DataGenerator.base64Characters().genBytes(_128KiB);
+      try (WritableByteChannel writer = recoveryFile.writer()) {
+        writer.write(ByteBuffer.wrap(bytes));
+      }
+
+      ByteArrayOutputStream baos = new ByteArrayOutputStream();
+      try (SeekableByteChannel r = recoveryFile.reader();
+          WritableByteChannel w = Channels.newChannel(baos)) {
+        long copy = ByteStreams.copy(r, w);
+        assertThat(copy).isEqualTo(_128KiB);
+      }
+
+      assertThat(baos.toByteArray()).isEqualTo(bytes);
+    }
+
+    try (Stream<Path> stream = Files.list(tempDir)) {
+      boolean b = stream.anyMatch(Objects::nonNull);
+      assertThat(b).isFalse();
+    }
+  }
+
+  @Test
+  public void argValidation_nonEmpty() {
+    IllegalArgumentException iae =
+        assertThrows(
+            IllegalArgumentException.class, () -> RecoveryFileManager.of(ImmutableList.of()));
+
+    assertThat(iae).hasMessageThat().isNotEmpty();
+  }
+
+  @Test
+  public void argValidation_fileInsteadOfDirectory() throws IOException {
+    Path tempDir = temporaryFolder.newFile(testName.getMethodName()).toPath();
+
+    IllegalArgumentException iae =
+        assertThrows(
+            IllegalArgumentException.class,
+            () -> RecoveryFileManager.of(ImmutableList.of(tempDir)));
+
+    assertThat(iae).hasMessageThat().isNotEmpty();
+  }
+
+  @Test
+  public void argValidation_directoryDoesNotExistIsCreated() throws IOException {
+    Path tempDir = temporaryFolder.newFolder(testName.getMethodName()).toPath();
+
+    Path subPathA = tempDir.resolve("sub/path/a");
+
+    assertThat(Files.exists(subPathA)).isFalse();
+    RecoveryFileManager rfm = RecoveryFileManager.of(ImmutableList.of(subPathA));
+    assertThat(Files.exists(subPathA)).isTrue();
+  }
+
+  @Test
+  public void fileAssignmentIsRoundRobin() throws IOException {
+    Path tempDir1 = temporaryFolder.newFolder(testName.getMethodName() + "1").toPath();
+    Path tempDir2 = temporaryFolder.newFolder(testName.getMethodName() + "2").toPath();
+    Path tempDir3 = temporaryFolder.newFolder(testName.getMethodName() + "3").toPath();
+    RecoveryFileManager rfm =
+        RecoveryFileManager.of(ImmutableList.of(tempDir1, tempDir2, tempDir3));
+
+    BlobInfo info1 = BlobInfo.newBuilder("bucket", "object1").build();
+    BlobInfo info2 = BlobInfo.newBuilder("bucket", "object2").build();
+    BlobInfo info3 = BlobInfo.newBuilder("bucket", "object3").build();
+    try (RecoveryFile recoveryFile1 = rfm.newRecoveryFile(info1);
+        RecoveryFile recoveryFile2 = rfm.newRecoveryFile(info2);
+        RecoveryFile recoveryFile3 = rfm.newRecoveryFile(info3)) {
+
+      ImmutableSet<Path> paths =
+          Stream.of(recoveryFile1, recoveryFile2, recoveryFile3)
+              .map(rf -> rf.unsafe().touch())
+              .map(Path::toAbsolutePath)
+              .collect(ImmutableSet.toImmutableSet());
+
+      ImmutableSet<Path> parentDirs =
+          Stream.of(recoveryFile1, recoveryFile2, recoveryFile3)
+              .map(RecoveryFile::getPath)
+              .map(Path::getParent)
+              .collect(ImmutableSet.toImmutableSet());
+
+      assertThat(paths).hasSize(3);
+      assertThat(parentDirs).isEqualTo(ImmutableSet.of(tempDir1, tempDir2, tempDir3));
+    }
+  }
+}
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/SerializationTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/SerializationTest.java
index 87b88a78..ae3a140f 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/SerializationTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/SerializationTest.java
@@ -170,35 +170,47 @@ public class SerializationTest extends BaseSerializationTest {
             .add(UnifiedOpts.md5MatchExtractor())
             .build();
 
-    return new Serializable[] {
-      ACL_DOMAIN,
-      ACL_GROUP,
-      ACL_PROJECT_,
-      ACL_USER,
-      ACL_RAW,
-      ACL,
-      BLOB_INFO,
-      BLOB,
-      BUCKET_INFO,
-      BUCKET,
-      ORIGIN,
-      CORS,
-      PAGE_RESULT,
-      BLOB_LIST_OPTIONS,
-      BLOB_SOURCE_OPTIONS,
-      BLOB_TARGET_OPTIONS,
-      BUCKET_LIST_OPTIONS,
-      BUCKET_SOURCE_OPTIONS,
-      BUCKET_TARGET_OPTIONS,
-      STORAGE_EXCEPTION,
-      optionsDefault1,
-      optionsDefault2,
-      optionsHttp1,
-      optionsHttp2,
-      optionsGrpc1,
-      optionsGrpc2,
-      serializableOpts
-    };
+    try {
+      GrpcStorageOptions grpcStorageOptionsBufferToTemp =
+          StorageOptions.grpc()
+              .setCredentials(NoCredentials.getInstance())
+              .setProjectId("project1")
+              .setBlobWriteSessionConfig(BlobWriteSessionConfigs.bufferToTempDirThenUpload())
+              .build();
+
+      return new Serializable[] {
+        ACL_DOMAIN,
+        ACL_GROUP,
+        ACL_PROJECT_,
+        ACL_USER,
+        ACL_RAW,
+        ACL,
+        BLOB_INFO,
+        BLOB,
+        BUCKET_INFO,
+        BUCKET,
+        ORIGIN,
+        CORS,
+        PAGE_RESULT,
+        BLOB_LIST_OPTIONS,
+        BLOB_SOURCE_OPTIONS,
+        BLOB_TARGET_OPTIONS,
+        BUCKET_LIST_OPTIONS,
+        BUCKET_SOURCE_OPTIONS,
+        BUCKET_TARGET_OPTIONS,
+        STORAGE_EXCEPTION,
+        optionsDefault1,
+        optionsDefault2,
+        optionsHttp1,
+        optionsHttp2,
+        optionsGrpc1,
+        optionsGrpc2,
+        serializableOpts,
+        grpcStorageOptionsBufferToTemp
+      };
+    } catch (IOException ioe) {
+      throw new AssertionError(ioe);
+    }
   }
 
   @Override
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/TestClock.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/TestClock.java
new file mode 100644
index 00000000..e37a4f02
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/TestClock.java
@@ -0,0 +1,62 @@
+/*
+ * Copyright 2023 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 java.time.Clock;
+import java.time.Duration;
+import java.time.Instant;
+import java.time.ZoneId;
+import java.util.function.UnaryOperator;
+
+final class TestClock extends Clock {
+
+  private final Instant begin;
+  private final UnaryOperator<Instant> next;
+
+  private Instant now;
+
+  private TestClock(Instant begin, UnaryOperator<Instant> next) {
+    this.begin = begin;
+    this.next = next;
+    this.now = begin;
+  }
+
+  @Override
+  public ZoneId getZone() {
+    throw new UnsupportedOperationException("TestClock.getZone()");
+  }
+
+  @Override
+  public Clock withZone(ZoneId zone) {
+    throw new UnsupportedOperationException("TestClock.withZone()");
+  }
+
+  @Override
+  public Instant instant() {
+    Instant ret = now;
+    now = next.apply(now);
+    return ret;
+  }
+
+  public static TestClock tickBy(Instant begin, Duration d) {
+    return of(begin, i -> i.plus(d));
+  }
+
+  public static TestClock of(Instant begin, UnaryOperator<Instant> next) {
+    return new TestClock(begin, next);
+  }
+}
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/ThroughputMovingWindowPropertyTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/ThroughputMovingWindowPropertyTest.java
new file mode 100644
index 00000000..181f51ae
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/ThroughputMovingWindowPropertyTest.java
@@ -0,0 +1,422 @@
+/*
+ * Copyright 2023 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 com.google.cloud.storage.ByteSizeConstants._5TiB;
+import static com.google.common.truth.Truth.assertWithMessage;
+import static java.time.Instant.EPOCH;
+import static java.time.Instant.ofEpochSecond;
+
+import com.google.common.base.MoreObjects;
+import com.google.common.collect.ImmutableList;
+import java.time.Duration;
+import java.time.Instant;
+import java.time.temporal.ChronoUnit;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.List;
+import java.util.stream.Collectors;
+import net.jqwik.api.Arbitraries;
+import net.jqwik.api.Arbitrary;
+import net.jqwik.api.Combinators;
+import net.jqwik.api.Example;
+import net.jqwik.api.ForAll;
+import net.jqwik.api.Property;
+import net.jqwik.api.Provide;
+import net.jqwik.api.Tuple;
+import net.jqwik.api.Tuple.Tuple1;
+import net.jqwik.time.api.DateTimes;
+import net.jqwik.time.api.Times;
+
+final class ThroughputMovingWindowPropertyTest {
+
+  private static final double TOLERANCE = 0.001;
+
+  @Example
+  void canned() {
+    test(CANNED_SCENARIO);
+  }
+
+  @Example
+  void twoEntriesSameTimeDifferentThroughput() {
+    Duration ms = Duration.ofMillis(1);
+    ScenarioTimeline scenario =
+        new ScenarioTimeline(
+            ms,
+            ImmutableList.of(
+                new TimelineEntry(EPOCH, Throughput.of(1, ms), 1000.0),
+                new TimelineEntry(EPOCH, Throughput.of(0, ms), 1000.0)));
+    test(scenario);
+  }
+
+  @Property
+  void test(@ForAll("Scenarios") ScenarioTimeline scenario) {
+    ThroughputMovingWindow window = ThroughputMovingWindow.of(scenario.d);
+    for (TimelineEntry timelineEntry : scenario.timelineEntries) {
+      window.add(timelineEntry.i, timelineEntry.t);
+      Throughput throughput = window.avg(timelineEntry.i);
+      assertWithMessage(timelineEntry.toString())
+          .that(throughput.toBps())
+          .isWithin(TOLERANCE)
+          .of(timelineEntry.expectedMovingAvgBytesPerSecond);
+    }
+  }
+
+  @Provide("Scenarios")
+  static Arbitrary<ScenarioTimeline> scenarioTimeline() {
+    return Times.durations()
+        .ofPrecision(ChronoUnit.MILLIS)
+        .between(Duration.ofMillis(1), Duration.ofMinutes(10))
+        .flatMap(
+            d ->
+                Combinators.combine(
+                        Arbitraries.just(d),
+                        // pick an instant, then generate 1 to 100 values between it and d * 3
+                        DateTimes.instants()
+                            .ofPrecision(ChronoUnit.MILLIS)
+                            .flatMap(
+                                i ->
+                                    DateTimes.instants()
+                                        .ofPrecision(ChronoUnit.MILLIS)
+                                        .between(i, i.plus(d.multipliedBy(3)))
+                                        .flatMap(
+                                            ii ->
+                                                Combinators.combine(
+                                                        Arbitraries.just(ii), throughput())
+                                                    .as(Tuple::of))
+                                        .list()
+                                        .ofMinSize(1)
+                                        .ofMaxSize(100)))
+                    .as(Tuple::of))
+        .map(ScenarioTimeline::create);
+  }
+
+  static Arbitrary<Throughput> throughput() {
+    return Times.durations()
+        .ofPrecision(ChronoUnit.MILLIS)
+        .between(Duration.ofMillis(1), Duration.ofMinutes(10))
+        .flatMap(d -> Arbitraries.longs().between(0, _5TiB).map(n -> Throughput.of(n, d)));
+  }
+
+  private static final class ScenarioTimeline {
+
+    private static final Comparator<Tuple.Tuple2<Instant, ?>> COMP =
+        Comparator.comparing(Tuple1::get1);
+    private final Duration d;
+    private final List<TimelineEntry> timelineEntries;
+
+    private ScenarioTimeline(Duration d, List<TimelineEntry> timelineEntries) {
+      this.d = d;
+      this.timelineEntries = timelineEntries;
+    }
+
+    @Override
+    public String toString() {
+      return MoreObjects.toStringHelper(this)
+          .add("d", d)
+          .add("timelineEntries", timelineEntries)
+          .toString();
+    }
+
+    static ScenarioTimeline create(
+        Tuple.Tuple2<Duration, List<Tuple.Tuple2<Instant, Throughput>>> tuples) {
+
+      Duration d = tuples.get1();
+      List<Tuple.Tuple2<Instant, Throughput>> pairs = tuples.get2();
+
+      List<Tuple.Tuple2<Instant, Throughput>> tmp =
+          pairs.stream().sorted(COMP).collect(Collectors.toList());
+
+      List<List<Tuple.Tuple2<Instant, Throughput>>> windows = new ArrayList<>();
+      int last = tmp.size() - 1;
+      for (int i = last; i >= 0; i--) {
+        List<Tuple.Tuple2<Instant, Throughput>> window = new ArrayList<>();
+        Tuple.Tuple2<Instant, Throughput> t = tmp.get(i);
+        window.add(t);
+        Instant min = t.get1().minus(d);
+        for (int j = i - 1; j >= 0; j--) {
+          Tuple.Tuple2<Instant, Throughput> r = tmp.get(j);
+          if (r.get1().isAfter(min)) {
+            window.add(r);
+          }
+        }
+        windows.add(ImmutableList.copyOf(window));
+      }
+
+      ImmutableList<TimelineEntry> timelineEntries =
+          windows.stream()
+              .map(
+                  w -> {
+                    Tuple.Tuple2<Instant, Throughput> max = w.get(0);
+                    Throughput reduce =
+                        w.stream()
+                            .map(Tuple.Tuple2::get2)
+                            .reduce(Throughput.zero(), Throughput::plus);
+                    return new TimelineEntry(
+                        max.get1(), max.get2(), Throughput.of(reduce.getNumBytes(), d).toBps());
+                  })
+              .collect(ImmutableList.toImmutableList());
+      return new ScenarioTimeline(d, timelineEntries.reverse());
+    }
+  }
+
+  private static final class TimelineEntry {
+    private final Instant i;
+    private final Throughput t;
+    private final double expectedMovingAvgBytesPerSecond;
+
+    private TimelineEntry(Instant i, Throughput t, double expectedMovingAvgBytesPerSecond) {
+      this.i = i;
+      this.t = t;
+      this.expectedMovingAvgBytesPerSecond = expectedMovingAvgBytesPerSecond;
+    }
+
+    @Override
+    public String toString() {
+      return MoreObjects.toStringHelper(this)
+          .add("i", i)
+          .add("t", t)
+          .add("tenSecMovingAvg", String.format("%,.03f", expectedMovingAvgBytesPerSecond))
+          .toString();
+    }
+  }
+
+  private static final ScenarioTimeline CANNED_SCENARIO =
+      new ScenarioTimeline(
+          Duration.ofSeconds(10),
+          ImmutableList.<TimelineEntry>builder()
+              .add(new TimelineEntry(ofEpochSecond(1), Throughput.bytesPerSecond(192), 19.2))
+              .add(new TimelineEntry(ofEpochSecond(2), Throughput.bytesPerSecond(1185), 137.7))
+              .add(new TimelineEntry(ofEpochSecond(3), Throughput.bytesPerSecond(1363), 274.))
+              .add(new TimelineEntry(ofEpochSecond(4), Throughput.bytesPerSecond(234), 297.4))
+              .add(new TimelineEntry(ofEpochSecond(5), Throughput.bytesPerSecond(1439), 441.3))
+              .add(new TimelineEntry(ofEpochSecond(6), Throughput.bytesPerSecond(1269), 568.2))
+              .add(new TimelineEntry(ofEpochSecond(7), Throughput.bytesPerSecond(692), 637.4))
+              .add(new TimelineEntry(ofEpochSecond(8), Throughput.bytesPerSecond(667), 704.1))
+              .add(new TimelineEntry(ofEpochSecond(9), Throughput.bytesPerSecond(1318), 835.9))
+              .add(new TimelineEntry(ofEpochSecond(10), Throughput.bytesPerSecond(1125), 948.4))
+              .add(new TimelineEntry(ofEpochSecond(11), Throughput.bytesPerSecond(1124), 1041.6))
+              .add(new TimelineEntry(ofEpochSecond(12), Throughput.bytesPerSecond(3), 923.4))
+              .add(new TimelineEntry(ofEpochSecond(13), Throughput.bytesPerSecond(185), 805.6))
+              .add(new TimelineEntry(ofEpochSecond(14), Throughput.bytesPerSecond(726), 854.8))
+              .add(new TimelineEntry(ofEpochSecond(15), Throughput.bytesPerSecond(630), 773.9))
+              .add(new TimelineEntry(ofEpochSecond(16), Throughput.bytesPerSecond(874), 734.4))
+              .add(new TimelineEntry(ofEpochSecond(17), Throughput.bytesPerSecond(1401), 805.3))
+              .add(new TimelineEntry(ofEpochSecond(18), Throughput.bytesPerSecond(533), 791.9))
+              .add(new TimelineEntry(ofEpochSecond(19), Throughput.bytesPerSecond(446), 704.7))
+              .add(new TimelineEntry(ofEpochSecond(20), Throughput.bytesPerSecond(801), 672.3))
+              .add(new TimelineEntry(ofEpochSecond(21), Throughput.bytesPerSecond(61), 566.))
+              .add(new TimelineEntry(ofEpochSecond(22), Throughput.bytesPerSecond(1104), 676.1))
+              .add(new TimelineEntry(ofEpochSecond(23), Throughput.bytesPerSecond(972), 754.8))
+              .add(new TimelineEntry(ofEpochSecond(24), Throughput.bytesPerSecond(1310), 813.2))
+              .add(new TimelineEntry(ofEpochSecond(25), Throughput.bytesPerSecond(408), 791.))
+              .add(new TimelineEntry(ofEpochSecond(26), Throughput.bytesPerSecond(759), 779.5))
+              .add(new TimelineEntry(ofEpochSecond(27), Throughput.bytesPerSecond(674), 706.8))
+              .add(new TimelineEntry(ofEpochSecond(28), Throughput.bytesPerSecond(314), 684.9))
+              .add(new TimelineEntry(ofEpochSecond(29), Throughput.bytesPerSecond(1311), 771.4))
+              .add(new TimelineEntry(ofEpochSecond(30), Throughput.bytesPerSecond(449), 736.2))
+              .add(new TimelineEntry(ofEpochSecond(31), Throughput.bytesPerSecond(1273), 857.4))
+              .add(new TimelineEntry(ofEpochSecond(32), Throughput.bytesPerSecond(228), 769.8))
+              .add(new TimelineEntry(ofEpochSecond(33), Throughput.bytesPerSecond(605), 733.1))
+              .add(new TimelineEntry(ofEpochSecond(34), Throughput.bytesPerSecond(537), 655.8))
+              .add(new TimelineEntry(ofEpochSecond(35), Throughput.bytesPerSecond(1498), 764.8))
+              .add(new TimelineEntry(ofEpochSecond(36), Throughput.bytesPerSecond(694), 758.3))
+              .add(new TimelineEntry(ofEpochSecond(37), Throughput.bytesPerSecond(155), 706.4))
+              .add(new TimelineEntry(ofEpochSecond(38), Throughput.bytesPerSecond(983), 773.3))
+              .add(new TimelineEntry(ofEpochSecond(39), Throughput.bytesPerSecond(1359), 778.1))
+              .add(new TimelineEntry(ofEpochSecond(40), Throughput.bytesPerSecond(832), 816.4))
+              .add(new TimelineEntry(ofEpochSecond(41), Throughput.bytesPerSecond(1041), 793.2))
+              .add(new TimelineEntry(ofEpochSecond(42), Throughput.bytesPerSecond(1459), 916.3))
+              .add(new TimelineEntry(ofEpochSecond(43), Throughput.bytesPerSecond(1128), 968.6))
+              .add(new TimelineEntry(ofEpochSecond(44), Throughput.bytesPerSecond(1318), 1046.7))
+              .add(new TimelineEntry(ofEpochSecond(45), Throughput.bytesPerSecond(620), 958.9))
+              .add(new TimelineEntry(ofEpochSecond(46), Throughput.bytesPerSecond(1133), 1002.8))
+              .add(new TimelineEntry(ofEpochSecond(47), Throughput.bytesPerSecond(568), 1044.1))
+              .add(new TimelineEntry(ofEpochSecond(48), Throughput.bytesPerSecond(561), 1001.9))
+              .add(new TimelineEntry(ofEpochSecond(49), Throughput.bytesPerSecond(1483), 1014.3))
+              .add(new TimelineEntry(ofEpochSecond(50), Throughput.bytesPerSecond(1405), 1071.6))
+              .add(new TimelineEntry(ofEpochSecond(51), Throughput.bytesPerSecond(435), 1011.))
+              .add(new TimelineEntry(ofEpochSecond(52), Throughput.bytesPerSecond(664), 931.5))
+              .add(new TimelineEntry(ofEpochSecond(53), Throughput.bytesPerSecond(1330), 951.7))
+              .add(new TimelineEntry(ofEpochSecond(54), Throughput.bytesPerSecond(540), 873.9))
+              .add(new TimelineEntry(ofEpochSecond(55), Throughput.bytesPerSecond(847), 896.6))
+              .add(new TimelineEntry(ofEpochSecond(56), Throughput.bytesPerSecond(1231), 906.4))
+              .add(new TimelineEntry(ofEpochSecond(57), Throughput.bytesPerSecond(1331), 982.7))
+              .add(new TimelineEntry(ofEpochSecond(58), Throughput.bytesPerSecond(154), 942.))
+              .add(new TimelineEntry(ofEpochSecond(59), Throughput.bytesPerSecond(801), 873.8))
+              .add(new TimelineEntry(ofEpochSecond(60), Throughput.bytesPerSecond(499), 783.2))
+              .add(new TimelineEntry(ofEpochSecond(61), Throughput.bytesPerSecond(766), 816.3))
+              .add(new TimelineEntry(ofEpochSecond(62), Throughput.bytesPerSecond(1166), 866.5))
+              .add(new TimelineEntry(ofEpochSecond(63), Throughput.bytesPerSecond(1408), 874.3))
+              .add(new TimelineEntry(ofEpochSecond(64), Throughput.bytesPerSecond(1145), 934.8))
+              .add(new TimelineEntry(ofEpochSecond(65), Throughput.bytesPerSecond(433), 893.4))
+              .add(new TimelineEntry(ofEpochSecond(66), Throughput.bytesPerSecond(1256), 895.9))
+              .add(new TimelineEntry(ofEpochSecond(67), Throughput.bytesPerSecond(847), 847.5))
+              .add(new TimelineEntry(ofEpochSecond(68), Throughput.bytesPerSecond(1421), 974.2))
+              .add(new TimelineEntry(ofEpochSecond(69), Throughput.bytesPerSecond(347), 928.8))
+              .add(new TimelineEntry(ofEpochSecond(70), Throughput.bytesPerSecond(52), 884.1))
+              .add(new TimelineEntry(ofEpochSecond(71), Throughput.bytesPerSecond(19), 809.4))
+              .add(new TimelineEntry(ofEpochSecond(72), Throughput.bytesPerSecond(1191), 811.9))
+              .add(new TimelineEntry(ofEpochSecond(73), Throughput.bytesPerSecond(104), 681.5))
+              .add(new TimelineEntry(ofEpochSecond(74), Throughput.bytesPerSecond(640), 631.))
+              .add(new TimelineEntry(ofEpochSecond(75), Throughput.bytesPerSecond(535), 641.2))
+              .add(new TimelineEntry(ofEpochSecond(76), Throughput.bytesPerSecond(203), 535.9))
+              .add(new TimelineEntry(ofEpochSecond(77), Throughput.bytesPerSecond(51), 456.3))
+              .add(new TimelineEntry(ofEpochSecond(78), Throughput.bytesPerSecond(1117), 425.9))
+              .add(new TimelineEntry(ofEpochSecond(79), Throughput.bytesPerSecond(1390), 530.2))
+              .add(new TimelineEntry(ofEpochSecond(80), Throughput.bytesPerSecond(262), 551.2))
+              .add(new TimelineEntry(ofEpochSecond(81), Throughput.bytesPerSecond(5), 549.8))
+              .add(new TimelineEntry(ofEpochSecond(82), Throughput.bytesPerSecond(802), 510.9))
+              .add(new TimelineEntry(ofEpochSecond(83), Throughput.bytesPerSecond(529), 553.4))
+              .add(new TimelineEntry(ofEpochSecond(84), Throughput.bytesPerSecond(1261), 615.5))
+              .add(new TimelineEntry(ofEpochSecond(85), Throughput.bytesPerSecond(1192), 681.2))
+              .add(new TimelineEntry(ofEpochSecond(86), Throughput.bytesPerSecond(276), 688.5))
+              .add(new TimelineEntry(ofEpochSecond(87), Throughput.bytesPerSecond(457), 729.1))
+              .add(new TimelineEntry(ofEpochSecond(88), Throughput.bytesPerSecond(799), 697.3))
+              .add(new TimelineEntry(ofEpochSecond(89), Throughput.bytesPerSecond(443), 602.6))
+              .add(new TimelineEntry(ofEpochSecond(90), Throughput.bytesPerSecond(1281), 704.5))
+              .add(new TimelineEntry(ofEpochSecond(91), Throughput.bytesPerSecond(97), 713.7))
+              .add(new TimelineEntry(ofEpochSecond(92), Throughput.bytesPerSecond(895), 723.))
+              .add(new TimelineEntry(ofEpochSecond(93), Throughput.bytesPerSecond(1338), 803.9))
+              .add(new TimelineEntry(ofEpochSecond(94), Throughput.bytesPerSecond(554), 733.2))
+              .add(new TimelineEntry(ofEpochSecond(95), Throughput.bytesPerSecond(302), 644.2))
+              .add(new TimelineEntry(ofEpochSecond(96), Throughput.bytesPerSecond(518), 668.4))
+              .add(new TimelineEntry(ofEpochSecond(97), Throughput.bytesPerSecond(502), 672.9))
+              .add(new TimelineEntry(ofEpochSecond(98), Throughput.bytesPerSecond(517), 644.7))
+              .add(new TimelineEntry(ofEpochSecond(99), Throughput.bytesPerSecond(172), 617.6))
+              .add(new TimelineEntry(ofEpochSecond(100), Throughput.bytesPerSecond(909), 580.4))
+              .add(new TimelineEntry(ofEpochSecond(101), Throughput.bytesPerSecond(1233), 694.))
+              .add(new TimelineEntry(ofEpochSecond(102), Throughput.bytesPerSecond(189), 623.4))
+              .add(new TimelineEntry(ofEpochSecond(103), Throughput.bytesPerSecond(244), 514.))
+              .add(new TimelineEntry(ofEpochSecond(104), Throughput.bytesPerSecond(886), 547.2))
+              .add(new TimelineEntry(ofEpochSecond(105), Throughput.bytesPerSecond(796), 596.6))
+              .add(new TimelineEntry(ofEpochSecond(106), Throughput.bytesPerSecond(1072), 652.))
+              .add(new TimelineEntry(ofEpochSecond(107), Throughput.bytesPerSecond(602), 662.))
+              .add(new TimelineEntry(ofEpochSecond(108), Throughput.bytesPerSecond(507), 661.))
+              .add(new TimelineEntry(ofEpochSecond(109), Throughput.bytesPerSecond(432), 687.))
+              .add(new TimelineEntry(ofEpochSecond(110), Throughput.bytesPerSecond(661), 662.2))
+              .add(new TimelineEntry(ofEpochSecond(111), Throughput.bytesPerSecond(1085), 647.4))
+              .add(new TimelineEntry(ofEpochSecond(112), Throughput.bytesPerSecond(157), 644.2))
+              .add(new TimelineEntry(ofEpochSecond(113), Throughput.bytesPerSecond(529), 672.7))
+              .add(new TimelineEntry(ofEpochSecond(114), Throughput.bytesPerSecond(31), 587.2))
+              .add(new TimelineEntry(ofEpochSecond(115), Throughput.bytesPerSecond(464), 554.))
+              .add(new TimelineEntry(ofEpochSecond(116), Throughput.bytesPerSecond(1301), 576.9))
+              .add(new TimelineEntry(ofEpochSecond(117), Throughput.bytesPerSecond(787), 595.4))
+              .add(new TimelineEntry(ofEpochSecond(118), Throughput.bytesPerSecond(908), 635.5))
+              .add(new TimelineEntry(ofEpochSecond(119), Throughput.bytesPerSecond(1316), 723.9))
+              .add(new TimelineEntry(ofEpochSecond(120), Throughput.bytesPerSecond(764), 734.2))
+              .add(new TimelineEntry(ofEpochSecond(121), Throughput.bytesPerSecond(1391), 764.8))
+              .add(new TimelineEntry(ofEpochSecond(122), Throughput.bytesPerSecond(819), 831.))
+              .add(new TimelineEntry(ofEpochSecond(123), Throughput.bytesPerSecond(219), 800.))
+              .add(new TimelineEntry(ofEpochSecond(124), Throughput.bytesPerSecond(601), 857.))
+              .add(new TimelineEntry(ofEpochSecond(125), Throughput.bytesPerSecond(1238), 934.4))
+              .add(new TimelineEntry(ofEpochSecond(126), Throughput.bytesPerSecond(1392), 943.5))
+              .add(new TimelineEntry(ofEpochSecond(127), Throughput.bytesPerSecond(499), 914.7))
+              .add(new TimelineEntry(ofEpochSecond(128), Throughput.bytesPerSecond(1153), 939.2))
+              .add(new TimelineEntry(ofEpochSecond(129), Throughput.bytesPerSecond(1219), 929.5))
+              .add(new TimelineEntry(ofEpochSecond(130), Throughput.bytesPerSecond(519), 905.))
+              .add(new TimelineEntry(ofEpochSecond(131), Throughput.bytesPerSecond(337), 799.6))
+              .add(new TimelineEntry(ofEpochSecond(132), Throughput.bytesPerSecond(1065), 824.2))
+              .add(new TimelineEntry(ofEpochSecond(133), Throughput.bytesPerSecond(789), 881.2))
+              .add(new TimelineEntry(ofEpochSecond(134), Throughput.bytesPerSecond(32), 824.3))
+              .add(new TimelineEntry(ofEpochSecond(135), Throughput.bytesPerSecond(893), 789.8))
+              .add(new TimelineEntry(ofEpochSecond(136), Throughput.bytesPerSecond(1093), 759.9))
+              .add(new TimelineEntry(ofEpochSecond(137), Throughput.bytesPerSecond(1218), 831.8))
+              .add(new TimelineEntry(ofEpochSecond(138), Throughput.bytesPerSecond(159), 732.4))
+              .add(new TimelineEntry(ofEpochSecond(139), Throughput.bytesPerSecond(407), 651.2))
+              .add(new TimelineEntry(ofEpochSecond(140), Throughput.bytesPerSecond(615), 660.8))
+              .add(new TimelineEntry(ofEpochSecond(141), Throughput.bytesPerSecond(1392), 766.3))
+              .add(new TimelineEntry(ofEpochSecond(142), Throughput.bytesPerSecond(1431), 802.9))
+              .add(new TimelineEntry(ofEpochSecond(143), Throughput.bytesPerSecond(270), 751.))
+              .add(new TimelineEntry(ofEpochSecond(144), Throughput.bytesPerSecond(300), 777.8))
+              .add(new TimelineEntry(ofEpochSecond(145), Throughput.bytesPerSecond(1402), 828.7))
+              .add(new TimelineEntry(ofEpochSecond(146), Throughput.bytesPerSecond(308), 750.2))
+              .add(new TimelineEntry(ofEpochSecond(147), Throughput.bytesPerSecond(125), 640.9))
+              .add(new TimelineEntry(ofEpochSecond(148), Throughput.bytesPerSecond(467), 671.7))
+              .add(new TimelineEntry(ofEpochSecond(149), Throughput.bytesPerSecond(1339), 764.9))
+              .add(new TimelineEntry(ofEpochSecond(150), Throughput.bytesPerSecond(1146), 818.))
+              .add(new TimelineEntry(ofEpochSecond(151), Throughput.bytesPerSecond(765), 755.3))
+              .add(new TimelineEntry(ofEpochSecond(152), Throughput.bytesPerSecond(649), 677.1))
+              .add(new TimelineEntry(ofEpochSecond(153), Throughput.bytesPerSecond(1318), 781.9))
+              .add(new TimelineEntry(ofEpochSecond(154), Throughput.bytesPerSecond(199), 771.8))
+              .add(new TimelineEntry(ofEpochSecond(155), Throughput.bytesPerSecond(923), 723.9))
+              .add(new TimelineEntry(ofEpochSecond(156), Throughput.bytesPerSecond(430), 736.1))
+              .add(new TimelineEntry(ofEpochSecond(157), Throughput.bytesPerSecond(158), 739.4))
+              .add(new TimelineEntry(ofEpochSecond(158), Throughput.bytesPerSecond(187), 711.4))
+              .add(new TimelineEntry(ofEpochSecond(159), Throughput.bytesPerSecond(442), 621.7))
+              .add(new TimelineEntry(ofEpochSecond(160), Throughput.bytesPerSecond(82), 515.3))
+              .add(new TimelineEntry(ofEpochSecond(161), Throughput.bytesPerSecond(951), 533.9))
+              .add(new TimelineEntry(ofEpochSecond(162), Throughput.bytesPerSecond(976), 566.6))
+              .add(new TimelineEntry(ofEpochSecond(163), Throughput.bytesPerSecond(1371), 571.9))
+              .add(new TimelineEntry(ofEpochSecond(164), Throughput.bytesPerSecond(547), 606.7))
+              .add(new TimelineEntry(ofEpochSecond(165), Throughput.bytesPerSecond(370), 551.4))
+              .add(new TimelineEntry(ofEpochSecond(166), Throughput.bytesPerSecond(247), 533.1))
+              .add(new TimelineEntry(ofEpochSecond(167), Throughput.bytesPerSecond(660), 583.3))
+              .add(new TimelineEntry(ofEpochSecond(168), Throughput.bytesPerSecond(1222), 686.8))
+              .add(new TimelineEntry(ofEpochSecond(169), Throughput.bytesPerSecond(130), 655.6))
+              .add(new TimelineEntry(ofEpochSecond(170), Throughput.bytesPerSecond(512), 698.6))
+              .add(new TimelineEntry(ofEpochSecond(171), Throughput.bytesPerSecond(873), 690.8))
+              .add(new TimelineEntry(ofEpochSecond(172), Throughput.bytesPerSecond(18), 595.))
+              .add(new TimelineEntry(ofEpochSecond(173), Throughput.bytesPerSecond(817), 539.6))
+              .add(new TimelineEntry(ofEpochSecond(174), Throughput.bytesPerSecond(1090), 593.9))
+              .add(new TimelineEntry(ofEpochSecond(175), Throughput.bytesPerSecond(1201), 677.))
+              .add(new TimelineEntry(ofEpochSecond(176), Throughput.bytesPerSecond(1046), 756.9))
+              .add(new TimelineEntry(ofEpochSecond(177), Throughput.bytesPerSecond(1075), 798.4))
+              .add(new TimelineEntry(ofEpochSecond(178), Throughput.bytesPerSecond(679), 744.1))
+              .add(new TimelineEntry(ofEpochSecond(179), Throughput.bytesPerSecond(1043), 835.4))
+              .add(new TimelineEntry(ofEpochSecond(180), Throughput.bytesPerSecond(1206), 904.8))
+              .add(new TimelineEntry(ofEpochSecond(181), Throughput.bytesPerSecond(701), 887.6))
+              .add(new TimelineEntry(ofEpochSecond(182), Throughput.bytesPerSecond(849), 970.7))
+              .add(new TimelineEntry(ofEpochSecond(183), Throughput.bytesPerSecond(457), 934.7))
+              .add(new TimelineEntry(ofEpochSecond(184), Throughput.bytesPerSecond(400), 865.7))
+              .add(new TimelineEntry(ofEpochSecond(185), Throughput.bytesPerSecond(1157), 861.3))
+              .add(new TimelineEntry(ofEpochSecond(186), Throughput.bytesPerSecond(235), 780.2))
+              .add(new TimelineEntry(ofEpochSecond(187), Throughput.bytesPerSecond(525), 725.2))
+              .add(new TimelineEntry(ofEpochSecond(188), Throughput.bytesPerSecond(1415), 798.8))
+              .add(new TimelineEntry(ofEpochSecond(189), Throughput.bytesPerSecond(796), 774.1))
+              .add(new TimelineEntry(ofEpochSecond(190), Throughput.bytesPerSecond(428), 696.3))
+              .add(new TimelineEntry(ofEpochSecond(191), Throughput.bytesPerSecond(417), 667.9))
+              .add(new TimelineEntry(ofEpochSecond(192), Throughput.bytesPerSecond(436), 626.6))
+              .add(new TimelineEntry(ofEpochSecond(193), Throughput.bytesPerSecond(781), 659.))
+              .add(new TimelineEntry(ofEpochSecond(194), Throughput.bytesPerSecond(967), 715.7))
+              .add(new TimelineEntry(ofEpochSecond(195), Throughput.bytesPerSecond(398), 639.8))
+              .add(new TimelineEntry(ofEpochSecond(196), Throughput.bytesPerSecond(501), 666.4))
+              .add(new TimelineEntry(ofEpochSecond(197), Throughput.bytesPerSecond(691), 683.))
+              .add(new TimelineEntry(ofEpochSecond(198), Throughput.bytesPerSecond(1492), 690.7))
+              .add(new TimelineEntry(ofEpochSecond(199), Throughput.bytesPerSecond(1493), 760.4))
+              .add(new TimelineEntry(ofEpochSecond(200), Throughput.bytesPerSecond(5), 718.1))
+              .add(new TimelineEntry(ofEpochSecond(201), Throughput.bytesPerSecond(679), 744.3))
+              .add(new TimelineEntry(ofEpochSecond(202), Throughput.bytesPerSecond(1027), 803.4))
+              .add(new TimelineEntry(ofEpochSecond(203), Throughput.bytesPerSecond(170), 742.3))
+              .add(new TimelineEntry(ofEpochSecond(204), Throughput.bytesPerSecond(261), 671.7))
+              .add(new TimelineEntry(ofEpochSecond(205), Throughput.bytesPerSecond(309), 662.8))
+              .add(new TimelineEntry(ofEpochSecond(206), Throughput.bytesPerSecond(1483), 761.))
+              .add(new TimelineEntry(ofEpochSecond(207), Throughput.bytesPerSecond(1154), 807.3))
+              .add(new TimelineEntry(ofEpochSecond(208), Throughput.bytesPerSecond(857), 743.8))
+              .add(new TimelineEntry(ofEpochSecond(209), Throughput.bytesPerSecond(792), 673.7))
+              .add(new TimelineEntry(ofEpochSecond(210), Throughput.bytesPerSecond(819), 755.1))
+              .add(new TimelineEntry(ofEpochSecond(211), Throughput.bytesPerSecond(763), 763.5))
+              .add(new TimelineEntry(ofEpochSecond(212), Throughput.bytesPerSecond(386), 699.4))
+              .add(new TimelineEntry(ofEpochSecond(213), Throughput.bytesPerSecond(789), 761.3))
+              .add(new TimelineEntry(ofEpochSecond(214), Throughput.bytesPerSecond(1432), 878.4))
+              .add(new TimelineEntry(ofEpochSecond(215), Throughput.bytesPerSecond(205), 868.))
+              .add(new TimelineEntry(ofEpochSecond(216), Throughput.bytesPerSecond(905), 810.2))
+              .add(new TimelineEntry(ofEpochSecond(217), Throughput.bytesPerSecond(1290), 823.8))
+              .add(new TimelineEntry(ofEpochSecond(218), Throughput.bytesPerSecond(639), 802.))
+              .add(new TimelineEntry(ofEpochSecond(219), Throughput.bytesPerSecond(1246), 847.4))
+              .build());
+}
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/ThroughputSinkTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/ThroughputSinkTest.java
new file mode 100644
index 00000000..a7840f45
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/ThroughputSinkTest.java
@@ -0,0 +1,262 @@
+/*
+ * Copyright 2023 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 com.google.cloud.storage.TestUtils.assertAll;
+import static com.google.common.truth.Truth.assertThat;
+
+import com.google.cloud.storage.ThroughputSink.Record;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.WritableByteChannel;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+import org.junit.Assert;
+import org.junit.Test;
+
+public final class ThroughputSinkTest {
+
+  @Test
+  public void tee_record() {
+    AtomicReference<Record> r1 = new AtomicReference<>(null);
+    AtomicReference<Record> r2 = new AtomicReference<>(null);
+    ThroughputSink test =
+        ThroughputSink.tee(
+            new AbstractThroughputSink() {
+              @Override
+              public void recordThroughput(Record r) {
+                r1.compareAndSet(null, r);
+              }
+            },
+            new AbstractThroughputSink() {
+              @Override
+              public void recordThroughput(Record r) {
+                r2.compareAndSet(null, r);
+              }
+            });
+
+    Record expected = Record.of(10, Instant.EPOCH, Instant.ofEpochSecond(1), false);
+    test.recordThroughput(expected);
+
+    assertThat(r1.get()).isEqualTo(expected);
+    assertThat(r2.get()).isEqualTo(expected);
+  }
+
+  @Test
+  public void tee_decorate() throws Exception {
+    AtomicReference<ByteBuffer> b1 = new AtomicReference<>(null);
+    AtomicReference<ByteBuffer> b2 = new AtomicReference<>(null);
+    AtomicReference<ByteBuffer> b3 = new AtomicReference<>(null);
+    ThroughputSink test =
+        ThroughputSink.tee(
+            new AbstractThroughputSink() {
+              @Override
+              public WritableByteChannel decorate(WritableByteChannel wbc) {
+                return new WritableByteChannel() {
+                  @Override
+                  public int write(ByteBuffer src) throws IOException {
+                    b1.compareAndSet(null, src.duplicate());
+                    return wbc.write(src);
+                  }
+
+                  @Override
+                  public boolean isOpen() {
+                    return wbc.isOpen();
+                  }
+
+                  @Override
+                  public void close() throws IOException {
+                    wbc.close();
+                  }
+                };
+              }
+            },
+            new AbstractThroughputSink() {
+              @Override
+              public WritableByteChannel decorate(WritableByteChannel wbc) {
+                return new WritableByteChannel() {
+                  @Override
+                  public int write(ByteBuffer src) throws IOException {
+                    ByteBuffer duplicate = src.duplicate();
+                    duplicate.position(src.limit());
+                    b2.compareAndSet(null, duplicate);
+                    return wbc.write(src);
+                  }
+
+                  @Override
+                  public boolean isOpen() {
+                    return wbc.isOpen();
+                  }
+
+                  @Override
+                  public void close() throws IOException {
+                    wbc.close();
+                  }
+                };
+              }
+            });
+
+    AtomicBoolean callIsOpen = new AtomicBoolean(false);
+    AtomicBoolean callClose = new AtomicBoolean(false);
+    WritableByteChannel anon =
+        new WritableByteChannel() {
+          @Override
+          public int write(ByteBuffer src) {
+            int remaining = src.remaining();
+            src.position(src.limit());
+            b3.compareAndSet(null, src);
+            return remaining;
+          }
+
+          @Override
+          public boolean isOpen() {
+            callIsOpen.compareAndSet(false, true);
+            return true;
+          }
+
+          @Override
+          public void close() {
+            callClose.compareAndSet(false, true);
+          }
+        };
+
+    byte[] bytes = DataGenerator.base64Characters().genBytes(16);
+
+    ByteBuffer expected1 = ByteBuffer.wrap(bytes);
+    ByteBuffer expected2 = ByteBuffer.wrap(bytes);
+    expected2.position(16);
+
+    ByteBuffer buf = ByteBuffer.wrap(bytes);
+    try (WritableByteChannel decorated = test.decorate(anon)) {
+      if (decorated.isOpen()) {
+        decorated.write(buf);
+      }
+    }
+
+    assertAll(
+        () -> assertThat(b1.get()).isEqualTo(expected1),
+        () -> assertThat(b2.get()).isEqualTo(expected2),
+        () -> assertThat(b3.get()).isSameInstanceAs(buf),
+        () -> assertThat(b3.get().hasRemaining()).isFalse(),
+        () -> assertThat(callIsOpen.get()).isTrue(),
+        () -> assertThat(callClose.get()).isTrue());
+  }
+
+  @Test
+  public void computeThroughput_noError() throws IOException {
+    // create a clock that will start at Epoch UTC, and will tick in one second increments
+    TestClock clock = TestClock.tickBy(Instant.EPOCH, Duration.ofSeconds(1));
+    AtomicReference<Record> actual = new AtomicReference<>(null);
+
+    ThroughputSink.computeThroughput(
+        clock,
+        new AbstractThroughputSink() {
+          @Override
+          public void recordThroughput(Record r) {
+            actual.compareAndSet(null, r);
+          }
+        },
+        300,
+        () -> {});
+
+    Record expected = Record.of(300, Instant.EPOCH, Instant.ofEpochSecond(1), false);
+    assertThat(actual.get()).isEqualTo(expected);
+  }
+
+  @Test
+  public void computeThroughput_ioError() {
+    // create a clock that will start at Epoch UTC, and will tick in one second increments
+    TestClock clock = TestClock.tickBy(Instant.EPOCH, Duration.ofSeconds(1));
+    AtomicReference<Record> actual = new AtomicReference<>(null);
+
+    IOException ioException =
+        Assert.assertThrows(
+            IOException.class,
+            () ->
+                ThroughputSink.computeThroughput(
+                    clock,
+                    new AbstractThroughputSink() {
+                      @Override
+                      public void recordThroughput(Record r) {
+                        actual.compareAndSet(null, r);
+                      }
+                    },
+                    300,
+                    () -> {
+                      throw new IOException("kablamo!");
+                    }));
+
+    Record expected = Record.of(300, Instant.EPOCH, Instant.ofEpochSecond(1), true);
+    assertThat(actual.get()).isEqualTo(expected);
+
+    assertThat(ioException).hasMessageThat().isEqualTo("kablamo!");
+  }
+
+  @Test
+  public void windowed() throws IOException {
+    // create a clock that will start at Epoch UTC, and will tick in one second increments
+    TestClock clock = TestClock.tickBy(Instant.EPOCH, Duration.ofSeconds(1));
+
+    AtomicReference<ByteBuffer> b3 = new AtomicReference<>(null);
+    WritableByteChannel anon =
+        new WritableByteChannel() {
+          @Override
+          public int write(ByteBuffer src) {
+            int remaining = src.remaining();
+            src.position(src.limit());
+            b3.compareAndSet(null, src);
+            return remaining;
+          }
+
+          @Override
+          public boolean isOpen() {
+            return true;
+          }
+
+          @Override
+          public void close() {}
+        };
+
+    Duration windowDuration = Duration.ofMinutes(1);
+    ThroughputMovingWindow window = ThroughputMovingWindow.of(windowDuration);
+    ThroughputSink sink = ThroughputSink.windowed(window, clock);
+
+    int numBytes = 120;
+    ByteBuffer buf = DataGenerator.base64Characters().genByteBuffer(numBytes);
+    try (WritableByteChannel decorated = sink.decorate(anon)) {
+      decorated.write(buf);
+    }
+
+    Throughput avg = window.avg(clock.instant());
+
+    assertThat(avg).isEqualTo(Throughput.of(numBytes, windowDuration));
+    assertThat(avg).isEqualTo(Throughput.bytesPerSecond(2));
+  }
+
+  private abstract static class AbstractThroughputSink implements ThroughputSink {
+
+    @Override
+    public void recordThroughput(Record r) {}
+
+    @Override
+    public WritableByteChannel decorate(WritableByteChannel wbc) {
+      return null;
+    }
+  }
+}
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/ThroughputTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/ThroughputTest.java
new file mode 100644
index 00000000..e44d4c1d
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/ThroughputTest.java
@@ -0,0 +1,35 @@
+/*
+ * Copyright 2023 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 com.google.common.truth.Truth.assertThat;
+import static java.time.Duration.ofSeconds;
+
+import org.junit.Test;
+
+public final class ThroughputTest {
+
+  @Test
+  public void a() {
+    assertThat(Throughput.bytesPerSecond(1).toBps()).isEqualTo(1);
+  }
+
+  @Test
+  public void b() {
+    assertThat(Throughput.of(10, ofSeconds(10)).toBps()).isEqualTo(1);
+  }
+}
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBlobWriteSessionTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBlobWriteSessionTest.java
index 1bb24fb9..1113d023 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBlobWriteSessionTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITBlobWriteSessionTest.java
@@ -61,6 +61,18 @@ public final class ITBlobWriteSessionTest {
     doTest(storage);
   }
 
+  @Test
+  public void bufferToTempDirThenUpload() throws Exception {
+    GrpcStorageOptions options =
+        ((GrpcStorageOptions) storage.getOptions())
+            .toBuilder()
+            .setBlobWriteSessionConfig(BlobWriteSessionConfigs.bufferToTempDirThenUpload())
+            .build();
+    try (Storage s = options.getService()) {
+      doTest(s);
+    }
+  }
+
   @Test
   public void overrideDefaultBufferSize() throws Exception {
     GrpcStorageOptions options =