Repository: googleapis/java-storage
Pull request #2138 (2023-07-24 22:05:05)

chore: introduce Throughput and ThroughputSink
_Pre-work_

In some upcoming work we want to be able to keep a running window of throughput performance in order to provide improved throughput. This PR introduces a new utility class to model and compute throughput, and the concept of a ThroughputSink which values can be appended to.




Patch:
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/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/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/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..5ef6e37d
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/ThroughputSink.java
@@ -0,0 +1,251 @@
+/*
+ * 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);
+  }
+
+  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);
+    }
+  }
+
+  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();
+    }
+  }
+
+  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));
+    }
+  }
+
+  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);
+    }
+  }
+}
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;
+    }
   }
 }


Test patch:
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/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);
+  }
+}