Repository: googleapis/java-storage
Pull request #2194 (2023-09-06 18:06:01)

feat: add new JournalingBlobWriteSessionConfig usable with gRPC transport
### Overview

New BlobWriteSessionConfig that will use disk to "Journal" data before transmitting to GCS.

By journaling the data to disk we can rewind to an arbitrary offset in the case of failure, while being optimistic and keeping the stream open to gcs.

General flow of data into this new WritableByteChannel:
1. #write(ByteBuffer)
2. Write contents of ByteBuffer to recovery file on disk
3. force a flush to disk
4. transmit contents of ByteBuffer to GCS leaving stream open
5. return

If the stream to gcs is interrupted with a retriable error
1. query the offset that was successfully committed to GCS
2. rewind our transmit context to the offset
3. Open the recovery file at offset from query
4. stream contents from recovery file to GCS leaving stream open
5. Once recovery file contents are transmitted to GCS return

### Benchmark results

#### Setup
1. c2d-standard-32 in us-east1 
   * with 4 local nvme sdd used for the recovery files 
   * default NIC, premium network tier 
   * debian 11 image
3. regional bucket in us-east1

#### Workload
generate a random file with size `1GiB..2GiB` upload random file to GCS using each of the following configurations:
1. DefaultBlobWriteSessionConfig with 16MiB chunk size
2. DefaultBlobWriteSessionConfig with 64MiB chunk size
4. BufferToDiskThenUpload with the following directories as buffer locations: /mnt/ssd1:/mnt/ssd2:/mnt/ssd3:/mnt/ssd4
5. JournalingBlobWriteSessionConfig with the following directories as journal locations: /mnt/ssd1:/mnt/ssd2:/mnt/ssd3:/mnt/ssd4

Run across `{1,2,4,8,16,32}` concurrent threads to evaluate contention and horizontal scaling.

#### Collected metrics
1. Generate random file
2. record begin instant
3. create new BlobWriteSession
4. BlobWriteSession#open()
6. Copy all bytes from random file to WritableByteChannel from 4
7. close the WritableByteChannel from 4
8. record end instant
9. report `objectSize,UploadStrategy,elapsedTimeUs,Status,ThreadCount`

#### Results summary

Throuhgput in MiB/s, grouped by ThreadCount and UploadStrategy
```
                             count     mean     std     min      50%      75%      90%      99%      max
ThreadCount UploadStrategy
1           PUT 16MiB       4341.0   79.941   8.752  21.599   80.218   85.628   90.710   99.635  106.627
            PUT 64MiB       4341.0  100.410  11.555  20.490  100.022  108.208  115.214  128.251  139.710
            journaling      4341.0  125.820  31.332  45.502  133.590  149.027  161.899  188.716  201.938
2           PUT 16MiB       4237.0   80.000   8.525  15.814   80.693   85.651   90.241   97.958  106.677
            PUT 64MiB       4237.0  101.062  11.030  55.813  101.049  108.007  115.114  127.299  135.149
            journaling      4237.0  125.010  29.761  43.207  131.827  147.362  159.425  182.441  209.000
4           PUT 16MiB       4411.0   79.708   8.357  40.691   80.600   85.567   89.586   95.533  103.506
            PUT 64MiB       4411.0  100.536   9.947  58.084  100.846  107.209  113.144  122.172  131.974
            journaling      4411.0  123.705  30.707  40.082  130.553  147.581  159.995  186.684  222.646
8           PUT 16MiB       4260.0   79.314   8.393   7.148   80.153   85.175   89.319   95.475  100.757
            PUT 64MiB       4260.0   99.913  10.438  60.685  100.450  107.144  112.551  122.409  132.130
            journaling      4260.0  122.931  30.261  42.747  130.306  146.098  158.005  184.798  203.696
16          PUT 16MiB       4473.0   77.735   8.091  24.149   78.483   83.123   87.092   95.740  106.176
            PUT 64MiB       4473.0   97.690   9.987  45.342   97.768  103.996  109.807  122.202  140.906
            journaling      4473.0  118.956  30.486  44.253  122.585  143.344  156.484  182.211  200.777
32          PUT 16MiB       4024.0   72.923   8.045  20.205   73.601   78.575   82.341   88.970  100.665
            PUT 64MiB       4024.0   93.151  10.030  20.913   93.506   99.748  105.297  116.163  128.284
            journaling      4024.0  104.557  30.965  11.785   98.794  129.923  146.747  174.618  200.303
```





Patch:
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 efaf569a..76296e9b 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
@@ -21,6 +21,7 @@ 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.channels.WritableByteChannel;
 import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.util.Collection;
@@ -28,6 +29,107 @@ import java.util.Collection;
 /**
  * Factory class to select and construct {@link BlobWriteSessionConfig}s.
  *
+ * <p>There are several strategies which can be used to upload a {@link Blob} to Google Cloud
+ * Storage. This class provides factories which allow you to select the appropriate strategy for
+ * your workload.
+ *
+ * <table>
+ *   <caption>Comparison of Strategies</caption>
+ *   <tr>
+ *     <th>Strategy</th>
+ *     <th>Factory Method(s)</th>
+ *     <th>Description</th>
+ *     <th>Retry Support</th>
+ *     <th>Transports Supported</th>
+ *     <th>Cloud Storage API used</th>
+ *     <th>Considerations</th>
+ *   </tr>
+ *   <tr>
+ *     <td>Default (Chunk based upload)</td>
+ *     <td>{@link #getDefault()}</td>
+ *     <td>
+ *       Buffer up to a configurable amount of bytes in memory, write to Cloud Storage when
+ *       full or close. Buffer size is configurable via
+ *       {@link DefaultBlobWriteSessionConfig#withChunkSize(int)}
+ *     </td>
+ *     <td>
+ *       Each chunk is retried up to the limitations specified in
+ *       {@link StorageOptions#getRetrySettings()}
+ *     </td>
+ *     <td>gRPC</td>
+ *     <td><a href="https://cloud.google.com/storage/docs/resumable-uploads">Resumable Upload</a></td>
+ *     <td>The network will only be used for the following operations:
+ *     <ol>
+ *       <li>Creating the Resumable Upload Session</li>
+ *       <li>Transmitting zero or more incremental chunks</li>
+ *       <li>Transmitting the final chunk and finalizing the Resumable Upload Session</li>
+ *       <li>
+ *         If any of the above are interrupted with a retryable error, the Resumable Upload Session
+ *         will be queried to reconcile client side state with Cloud Storage
+ *       </li>
+ *     </ol>
+ *     </td>
+ *   </tr>
+ *   <tr>
+ *     <td>Buffer to disk then upload</td>
+ *     <td>
+ *       <ul>
+ *         <li>{@link #bufferToDiskThenUpload(Path)}</li>
+ *         <li>{@link #bufferToDiskThenUpload(Collection) bufferToDiskThenUpload(Collection&lt;Path>)}</li>
+ *         <li>{@link #bufferToTempDirThenUpload()}</li>
+ *       </ul>
+ *     </td>
+ *     <td>
+ *       Buffer bytes to a temporary file on disk. On {@link WritableByteChannel#close() close()}
+ *       upload the entire files contents to Cloud Storage. Delete the temporary file.
+ *     </td>
+ *     <td>
+ *       Upload the file in the fewest number of RPC possible retrying within the limitations
+ *       specified in {@link StorageOptions#getRetrySettings()}
+ *     </td>
+ *     <td>gRPC</td>
+ *     <td><a href="https://cloud.google.com/storage/docs/resumable-uploads">Resumable Upload</a></td>
+ *     <td>
+ *       <ol>
+ *         <li>A Resumable Upload Session will be used to upload the file on disk.</li>
+ *         <li>
+ *           If the upload is interrupted with a retryable error, the Resumable Upload Session will
+ *           be queried to restart the upload from Cloud Storage's last received byte
+ *         </li>
+ *       </ol>
+ *     </td>
+ *   </tr>
+ *   <tr>
+ *     <td>Journal to disk while uploading</td>
+ *     <td>{@link #journaling(Collection) journaling(Collection&lt;Path>)}</td>
+ *     <td>
+ *       Create a Resumable Upload Session, before transmitting bytes to Cloud Storage write
+ *       to a recovery file on disk. If the stream to Cloud Storage is interrupted with a
+ *       retryable error query the offset of the Resumable Upload Session, then open the recovery
+ *       file from the offset and transmit the bytes to Cloud Storage.
+ *     </td>
+ *     <td>gRPC</td>
+ *     <td><a href="https://cloud.google.com/storage/docs/resumable-uploads">Resumable Upload</a></td>
+ *     <td>
+ *       <ol>
+ *         <li>
+ *           The stream to Cloud Storage will be held open until a) the write is complete
+ *           b) the stream is interrupted
+ *         </li>
+ *         <li>
+ *           Because the bytes are journaled to disk, the upload to Cloud Storage can only
+ *           be as fast as the disk.
+ *         </li>
+ *         <li>
+ *           The use of <a href="https://cloud.google.com/compute/docs/disks/local-ssd#nvme">Compute
+ *           Engine Local NVMe SSD</a> is strongly encouraged compared to Compute Engine Persistent
+ *           Disk.
+ *         </li>
+ *       </ol>
+ *     </td>
+ *   </tr>
+ * </table>
+ *
  * @see BlobWriteSessionConfig
  * @see GrpcStorageOptions.Builder#setBlobWriteSessionConfig(BlobWriteSessionConfig)
  * @see Storage#blobWriteSession(BlobInfo, BlobWriteOption...)
@@ -56,11 +158,11 @@ public final class BlobWriteSessionConfigs {
    * 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.
+   * <p>Once the file on disk is closed, the entire file will then be uploaded to Cloud Storage.
    *
    * @see Storage#blobWriteSession(BlobInfo, BlobWriteOption...)
    * @see GrpcStorageOptions.Builder#setBlobWriteSessionConfig(BlobWriteSessionConfig)
+   * @since 2.26.0 This new api is in preview and is subject to breaking changes.
    */
   @BetaApi
   public static BlobWriteSessionConfig bufferToTempDirThenUpload() throws IOException {
@@ -72,11 +174,11 @@ public final class BlobWriteSessionConfigs {
    * 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.
+   * <p>Once the file on disk is closed, the entire file will then be uploaded to Cloud Storage.
    *
    * @see Storage#blobWriteSession(BlobInfo, BlobWriteOption...)
    * @see GrpcStorageOptions.Builder#setBlobWriteSessionConfig(BlobWriteSessionConfig)
+   * @since 2.26.0 This new api is in preview and is subject to breaking changes.
    */
   @BetaApi
   public static BufferToDiskThenUpload bufferToDiskThenUpload(Path path) throws IOException {
@@ -87,18 +189,34 @@ public final class BlobWriteSessionConfigs {
    * 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>Once the file on disk is closed, the entire file will then be uploaded to 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)
+   * @since 2.26.0 This new api is in preview and is subject to breaking changes.
    */
   @BetaApi
   public static BufferToDiskThenUpload bufferToDiskThenUpload(Collection<Path> paths)
       throws IOException {
     return new BufferToDiskThenUpload(ImmutableList.copyOf(paths), false);
   }
+
+  /**
+   * Create a new {@link BlobWriteSessionConfig} which will journal writes to a temporary file under
+   * one of the specified {@code paths} before transmitting the bytes to 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)
+   * @since 2.27.0 This new api is in preview and is subject to breaking changes.
+   */
+  @BetaApi
+  public static JournalingBlobWriteSessionConfig journaling(Collection<Path> paths) {
+    return new JournalingBlobWriteSessionConfig(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
index fb20747a..430f30bc 100644
--- 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
@@ -22,6 +22,7 @@ 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.RecoveryFileManager.RecoveryVolumeSinkFactory;
 import com.google.cloud.storage.Storage.BlobWriteOption;
 import com.google.cloud.storage.UnifiedOpts.ObjectTargetOpt;
 import com.google.cloud.storage.UnifiedOpts.Opts;
@@ -96,8 +97,7 @@ public final class BufferToDiskThenUpload extends BlobWriteSessionConfig {
     return new Factory(recoveryFileManager, clock, gcs);
   }
 
-  private RecoveryFileManager.RecoverVolumeSinkFactory getRecoverVolumeSinkFactory(
-      Clock clock, Duration window) {
+  private RecoveryVolumeSinkFactory getRecoverVolumeSinkFactory(Clock clock, Duration window) {
     return path -> {
       ThroughputSink windowed = ThroughputSink.windowed(ThroughputMovingWindow.of(window), clock);
       if (includeLoggingSink) {
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/Buffers.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/Buffers.java
index a828523c..e793bcfe 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/Buffers.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/Buffers.java
@@ -16,8 +16,10 @@
 
 package com.google.cloud.storage;
 
+import java.io.IOException;
 import java.nio.Buffer;
 import java.nio.ByteBuffer;
+import java.nio.channels.ReadableByteChannel;
 import java.util.function.Consumer;
 
 /**
@@ -137,4 +139,19 @@ final class Buffers {
     } // else size is already aligned
     return alignedSize;
   }
+
+  static int fillFrom(ByteBuffer buf, ReadableByteChannel c) throws IOException {
+    int total = 0;
+    while (buf.hasRemaining()) {
+      int read = c.read(buf);
+      if (read != -1) {
+        total += read;
+      } else if (total == 0) {
+        return -1;
+      } else {
+        break;
+      }
+    }
+    return total;
+  }
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/ChunkSegmenter.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/ChunkSegmenter.java
index b891bb3b..50e3d7e6 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/ChunkSegmenter.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/ChunkSegmenter.java
@@ -50,6 +50,10 @@ final class ChunkSegmenter {
     return hasher;
   }
 
+  ChunkSegment[] segmentBuffer(ByteBuffer bb) {
+    return segmentBuffers(new ByteBuffer[] {bb}, 0, 1);
+  }
+
   /**
    * Given {@code bbs}, yield N segments, where each segment is at most {@code maxSegmentSize}
    * bytes.
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 9b4a3983..79e3759e 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
@@ -79,7 +79,8 @@ public final class DefaultBlobWriteSessionConfig extends BlobWriteSessionConfig
    *
    * <p><i>Default:</i> {@code 16777216 (16 MiB)}
    *
-   * @param chunkSize The number of bytes each chunk should be. Must be >= {@code 262144 (256 KiB)}
+   * @param chunkSize The number of bytes each chunk should be. Must be &gt;= {@code 262144 (256
+   *     KiB)}
    * @return The new instance
    * @see #getChunkSize()
    * @since 2.26.0 This new api is in preview and is subject to breaking changes.
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 f1b14d06..cdd964f8 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
@@ -145,7 +145,8 @@ final class DefaultBufferedWritableByteChannel implements BufferedWritableByteCh
           ByteBuffer slice = src.slice();
           Buffers.limit(slice, bufferRemaining);
           int write = channel.write(slice);
-          Buffers.position(src, srcPosition + write);
+          int newPosition = srcPosition + write;
+          Buffers.position(src, newPosition);
           bytesConsumed += write;
         }
       }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicWritableByteChannelSessionBuilder.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicWritableByteChannelSessionBuilder.java
index 80bca1c5..5aa47eb6 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicWritableByteChannelSessionBuilder.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicWritableByteChannelSessionBuilder.java
@@ -24,12 +24,15 @@ import com.google.api.core.InternalApi;
 import com.google.api.core.SettableApiFuture;
 import com.google.api.gax.retrying.ResultRetryAlgorithm;
 import com.google.api.gax.rpc.ClientStreamingCallable;
+import com.google.api.gax.rpc.UnaryCallable;
 import com.google.cloud.storage.ChannelSession.BufferedWriteSession;
 import com.google.cloud.storage.ChannelSession.UnbufferedWriteSession;
 import com.google.cloud.storage.Retrying.RetryingDependencies;
 import com.google.cloud.storage.UnbufferedWritableByteChannelSession.UnbufferedWritableByteChannel;
 import com.google.cloud.storage.WriteCtx.WriteObjectRequestBuilderFactory;
 import com.google.cloud.storage.WriteFlushStrategy.FlusherFactory;
+import com.google.storage.v2.QueryWriteStatusRequest;
+import com.google.storage.v2.QueryWriteStatusResponse;
 import com.google.storage.v2.ServiceConstants.Values;
 import com.google.storage.v2.WriteObjectRequest;
 import com.google.storage.v2.WriteObjectResponse;
@@ -106,6 +109,10 @@ final class GapicWritableByteChannelSessionBuilder {
     return new ResumableUploadBuilder();
   }
 
+  JournalingResumableUploadBuilder journaling() {
+    return new JournalingResumableUploadBuilder();
+  }
+
   /**
    * When constructing any of our channel sessions, there is always a {@link
    * GapicUnbufferedWritableByteChannel} at the bottom of it. This method creates a BiFunction which
@@ -332,4 +339,102 @@ final class GapicWritableByteChannelSessionBuilder {
       }
     }
   }
+
+  final class JournalingResumableUploadBuilder {
+
+    private RetryingDependencies deps;
+    private ResultRetryAlgorithm<?> alg;
+    private BufferHandle bufferHandle;
+    private BufferHandle recoveryBuffer;
+    private RecoveryFile recoveryFile;
+    private UnaryCallable<QueryWriteStatusRequest, QueryWriteStatusResponse> query;
+
+    JournalingResumableUploadBuilder() {
+      this.deps = RetryingDependencies.attemptOnce();
+      this.alg = Retrying.neverRetry();
+    }
+
+    JournalingResumableUploadBuilder withRetryConfig(
+        RetryingDependencies deps,
+        ResultRetryAlgorithm<?> alg,
+        UnaryCallable<QueryWriteStatusRequest, QueryWriteStatusResponse> query) {
+      this.deps = requireNonNull(deps, "deps must be non null");
+      this.alg = requireNonNull(alg, "alg must be non null");
+      this.query = requireNonNull(query, "query must be non null");
+      return this;
+    }
+
+    JournalingResumableUploadBuilder withBuffer(BufferHandle bufferHandle) {
+      this.bufferHandle = requireNonNull(bufferHandle, "bufferHandle must be non null");
+      return this;
+    }
+
+    JournalingResumableUploadBuilder withRecoveryBuffer(BufferHandle bufferHandle) {
+      this.recoveryBuffer = requireNonNull(bufferHandle, "bufferHandle must be non null");
+      return this;
+    }
+
+    JournalingResumableUploadBuilder withRecoveryFile(RecoveryFile recoveryFile) {
+      this.recoveryFile = requireNonNull(recoveryFile, "recoveryFile must be non null");
+      return this;
+    }
+
+    /**
+     * Set the Future which will contain the ResumableWrite information necessary to open the Write
+     * stream.
+     */
+    BuildableJournalingResumableUploadBuilder setStartAsync(
+        ApiFuture<WriteCtx<ResumableWrite>> start) {
+      requireNonNull(start, "start must be non null");
+      return new BuildableJournalingResumableUploadBuilder(start);
+    }
+
+    final class BuildableJournalingResumableUploadBuilder {
+      private final ApiFuture<WriteCtx<ResumableWrite>> start;
+
+      private BuildableJournalingResumableUploadBuilder(ApiFuture<WriteCtx<ResumableWrite>> start) {
+        this.start = start;
+      }
+
+      BufferedWritableByteChannelSession<WriteObjectResponse> build() {
+        return new BufferedWriteSession<>(
+            requireNonNull(start, "start must be non null"),
+            bindFunction()
+                .andThen(c -> new DefaultBufferedWritableByteChannel(bufferHandle, c))
+                .andThen(StorageByteChannels.writable()::createSynchronized));
+      }
+
+      private BiFunction<
+              WriteCtx<ResumableWrite>,
+              SettableApiFuture<WriteObjectResponse>,
+              UnbufferedWritableByteChannel>
+          bindFunction() {
+        // it is theoretically possible that the setter methods for the following variables could
+        // be called again between when this method is invoked and the resulting function is
+        // invoked.
+        // To ensure we are using the specified values at the point in time they are bound to the
+        // function read them into local variables which will be closed over rather than the class
+        // fields.
+        RetryingDependencies deps = JournalingResumableUploadBuilder.this.deps;
+        ResultRetryAlgorithm<?> alg = JournalingResumableUploadBuilder.this.alg;
+        BufferHandle recoveryBuffer = JournalingResumableUploadBuilder.this.recoveryBuffer;
+        RecoveryFile recoveryFile = JournalingResumableUploadBuilder.this.recoveryFile;
+        UnaryCallable<QueryWriteStatusRequest, QueryWriteStatusResponse> query =
+            JournalingResumableUploadBuilder.this.query;
+        ByteStringStrategy boundStrategy = byteStringStrategy;
+        Hasher boundHasher = hasher;
+        return (writeCtx, resultFuture) ->
+            new SyncAndUploadUnbufferedWritableByteChannel(
+                write,
+                query,
+                resultFuture,
+                new ChunkSegmenter(boundHasher, boundStrategy, Values.MAX_WRITE_CHUNK_BYTES_VALUE),
+                deps,
+                alg,
+                writeCtx,
+                recoveryFile,
+                recoveryBuffer);
+      }
+    }
+  }
 }
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 95ddad87..c9736992 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
@@ -1768,7 +1768,7 @@ final class GrpcStorageImpl extends BaseService<StorageOptions>
     return opts.readObjectRequest().apply(builder).build();
   }
 
-  private WriteObjectRequest getWriteObjectRequest(BlobInfo info, Opts<ObjectTargetOpt> opts) {
+  WriteObjectRequest getWriteObjectRequest(BlobInfo info, Opts<ObjectTargetOpt> opts) {
     Object object = codecs.blobInfo().encode(info);
     Object.Builder objectBuilder =
         object
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/JournalingBlobWriteSessionConfig.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/JournalingBlobWriteSessionConfig.java
new file mode 100644
index 00000000..c8cc0e3d
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/JournalingBlobWriteSessionConfig.java
@@ -0,0 +1,257 @@
+/*
+ * 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.ApiFutureCallback;
+import com.google.api.core.ApiFutures;
+import com.google.api.core.BetaApi;
+import com.google.api.core.InternalApi;
+import com.google.api.gax.grpc.GrpcCallContext;
+import com.google.api.gax.rpc.ApiExceptions;
+import com.google.cloud.storage.Conversions.Decoder;
+import com.google.cloud.storage.RecoveryFileManager.RecoveryVolumeSinkFactory;
+import com.google.cloud.storage.Storage.BlobWriteOption;
+import com.google.cloud.storage.ThroughputSink.Record;
+import com.google.cloud.storage.TransportCompatibility.Transport;
+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.ServiceConstants.Values;
+import com.google.storage.v2.WriteObjectResponse;
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.nio.channels.WritableByteChannel;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.time.Clock;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.ArrayList;
+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>If we have disk available we can checkpoint the contents of an object to disk before
+ * transmitting to GCS. The checkpointed data on disk allows arbitrary rewind in the case of failure
+ * but allows the upload to happen as soon as the checkpoint ack is complete.
+ *
+ * <p>Due to the details of how Resumable Upload Sessions are implemented in the GCS gRPC API this
+ * is possible. However, this approach will not work with the HTTP transports Resumable Upload
+ * Session spec.
+ *
+ * @see Storage#blobWriteSession(BlobInfo, BlobWriteOption...)
+ * @see GrpcStorageOptions.Builder#setBlobWriteSessionConfig(BlobWriteSessionConfig)
+ */
+@Immutable
+@BetaApi
+@TransportCompatibility(Transport.GRPC)
+public final class JournalingBlobWriteSessionConfig 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
+  JournalingBlobWriteSessionConfig(ImmutableList<Path> paths, boolean includeLoggingSink) {
+    this.paths = paths;
+    this.includeLoggingSink = includeLoggingSink;
+  }
+
+  @VisibleForTesting
+  @InternalApi
+  JournalingBlobWriteSessionConfig withIncludeLoggingSink() {
+    return new JournalingBlobWriteSessionConfig(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 RecoveryVolumeSinkFactory 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) {
+      if (storage instanceof GrpcStorageImpl) {
+        GrpcStorageImpl grpcStorage = (GrpcStorageImpl) storage;
+        RecoveryFile recoveryFile = recoveryFileManager.newRecoveryFile(info);
+        ApiFuture<ResumableWrite> f =
+            grpcStorage.startResumableWrite(
+                GrpcCallContext.createDefault(), grpcStorage.getWriteObjectRequest(info, opts));
+        ApiFuture<WriteCtx<ResumableWrite>> start =
+            ApiFutures.transform(f, WriteCtx::new, MoreExecutors.directExecutor());
+
+        BufferedWritableByteChannelSession<WriteObjectResponse> session =
+            ResumableMedia.gapic()
+                .write()
+                .byteChannel(grpcStorage.storageClient.writeObjectCallable())
+                .setHasher(Hasher.noop())
+                .setByteStringStrategy(ByteStringStrategy.copy())
+                .journaling()
+                .withRetryConfig(
+                    grpcStorage.getOptions(),
+                    grpcStorage.retryAlgorithmManager.idempotent(),
+                    grpcStorage.storageClient.queryWriteStatusCallable())
+                .withBuffer(BufferHandle.allocate(Values.MAX_WRITE_CHUNK_BYTES_VALUE))
+                .withRecoveryBuffer(BufferHandle.allocate(Values.MAX_WRITE_CHUNK_BYTES_VALUE))
+                .withRecoveryFile(recoveryFile)
+                .setStartAsync(start)
+                .build();
+
+        return new JournalingUpload<>(session, start, d);
+      } else {
+        return CrossTransportUtils.throwHttpJsonOnly(BlobWriteSessionConfigs.class, "journaling");
+      }
+    }
+
+    private final class JournalingUpload<WBC extends WritableByteChannel>
+        implements WritableByteChannelSession<WBC, BlobInfo> {
+
+      private final WritableByteChannelSession<WBC, WriteObjectResponse> session;
+      private final ApiFuture<WriteCtx<ResumableWrite>> start;
+      private final Decoder<WriteObjectResponse, BlobInfo> decoder;
+
+      public JournalingUpload(
+          WritableByteChannelSession<WBC, WriteObjectResponse> session,
+          ApiFuture<WriteCtx<ResumableWrite>> start,
+          Decoder<WriteObjectResponse, BlobInfo> decoder) {
+        this.session = session;
+        this.start = start;
+        this.decoder = decoder;
+      }
+
+      @Override
+      public ApiFuture<WBC> openAsync() {
+        // register a callback on the result future to record our throughput estimate
+        Instant begin = clock.instant();
+        ApiFutures.addCallback(
+            session.getResult(),
+            new ApiFutureCallback<WriteObjectResponse>() {
+              @Override
+              public void onFailure(Throwable t) {
+                Instant end = clock.instant();
+                // start MUST have completed in order for result to resolve, use the utility method
+                // to take care of the checked exception handling
+                WriteCtx<ResumableWrite> writeCtx =
+                    ApiExceptions.callAndTranslateApiException(start);
+                long totalSentBytes = writeCtx.getTotalSentBytes().get();
+                gcs.recordThroughput(Record.of(totalSentBytes, begin, end, true));
+              }
+
+              @Override
+              public void onSuccess(WriteObjectResponse result) {
+                Instant end = clock.instant();
+                long totalSentBytes = -1;
+                if (result.hasResource()) {
+                  totalSentBytes = result.getResource().getSize();
+                } else if (result.hasPersistedSize()) {
+                  totalSentBytes = result.getPersistedSize();
+                }
+                if (totalSentBytes > -1) {
+                  gcs.recordThroughput(Record.of(totalSentBytes, begin, end, false));
+                }
+              }
+            },
+            MoreExecutors.directExecutor());
+        return session.openAsync();
+      }
+
+      @Override
+      public ApiFuture<BlobInfo> getResult() {
+        return ApiFutures.transform(
+            session.getResult(), decoder::decode, MoreExecutors.directExecutor());
+      }
+    }
+  }
+}
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
index d399ea93..75884657 100644
--- 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
@@ -19,6 +19,8 @@ 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.FileChannel;
+import java.nio.channels.GatheringByteChannel;
 import java.nio.channels.SeekableByteChannel;
 import java.nio.channels.WritableByteChannel;
 import java.nio.file.Files;
@@ -33,18 +35,27 @@ import java.util.Set;
  * 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> writeOpsNew =
+      ImmutableSet.of(
+          StandardOpenOption.CREATE,
+          StandardOpenOption.WRITE,
+          StandardOpenOption.TRUNCATE_EXISTING);
+  private static final Set<OpenOption> writeOpsExisting =
+      ImmutableSet.of(
+          StandardOpenOption.CREATE, StandardOpenOption.WRITE, StandardOpenOption.APPEND);
   private static final Set<OpenOption> readOps = ImmutableSet.of(StandardOpenOption.READ);
 
   private final Path path;
   private final ThroughputSink throughputSink;
   private final Runnable onCloseCallback;
 
+  private boolean newFile;
+
   RecoveryFile(Path path, ThroughputSink throughputSink, Runnable onCloseCallback) {
     this.path = path;
     this.throughputSink = throughputSink;
     this.onCloseCallback = onCloseCallback;
+    this.newFile = true;
   }
 
   public Path getPath() {
@@ -52,6 +63,7 @@ final class RecoveryFile implements AutoCloseable {
   }
 
   public Path touch() throws IOException {
+    newFile = false;
     return Files.createFile(path);
   }
 
@@ -60,12 +72,26 @@ final class RecoveryFile implements AutoCloseable {
   }
 
   public WritableByteChannel writer() throws IOException {
-    return throughputSink.decorate(Files.newByteChannel(path, writeOps));
+    try {
+      return throughputSink.decorate(
+          Files.newByteChannel(path, newFile ? writeOpsNew : writeOpsExisting));
+    } finally {
+      newFile = false;
+    }
+  }
+
+  public GatheringByteChannel syncingChannel() throws IOException {
+    try {
+      return throughputSink.decorate(
+          new SyncingFileChannel(FileChannel.open(path, newFile ? writeOpsNew : writeOpsExisting)));
+    } finally {
+      newFile = false;
+    }
   }
 
   @Override
   public void close() throws IOException {
-    Files.delete(path);
+    Files.deleteIfExists(path);
     onCloseCallback.run();
   }
 
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
index 25303b5f..4926c899 100644
--- 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
@@ -66,7 +66,7 @@ final class RecoveryFileManager {
     return of(volumes, p -> ThroughputSink.nullSink());
   }
 
-  static RecoveryFileManager of(List<Path> volumes, RecoverVolumeSinkFactory factory)
+  static RecoveryFileManager of(List<Path> volumes, RecoveryVolumeSinkFactory factory)
       throws IOException {
     checkArgument(!volumes.isEmpty(), "At least one volume must be specified");
     checkArgument(
@@ -86,7 +86,7 @@ final class RecoveryFileManager {
   }
 
   @FunctionalInterface
-  interface RecoverVolumeSinkFactory {
+  interface RecoveryVolumeSinkFactory {
     ThroughputSink apply(Path p);
   }
 
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/SyncAndUploadUnbufferedWritableByteChannel.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/SyncAndUploadUnbufferedWritableByteChannel.java
new file mode 100644
index 00000000..4169b841
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/SyncAndUploadUnbufferedWritableByteChannel.java
@@ -0,0 +1,468 @@
+/*
+ * 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.WriteFlushStrategy.contextWithBucketName;
+
+import com.google.api.core.SettableApiFuture;
+import com.google.api.gax.grpc.GrpcCallContext;
+import com.google.api.gax.grpc.GrpcStatusCode;
+import com.google.api.gax.retrying.ResultRetryAlgorithm;
+import com.google.api.gax.retrying.TimedAttemptSettings;
+import com.google.api.gax.rpc.ApiException;
+import com.google.api.gax.rpc.ApiExceptionFactory;
+import com.google.api.gax.rpc.ApiExceptions;
+import com.google.api.gax.rpc.ApiStreamObserver;
+import com.google.api.gax.rpc.ClientStreamingCallable;
+import com.google.api.gax.rpc.UnaryCallable;
+import com.google.cloud.storage.ChunkSegmenter.ChunkSegment;
+import com.google.cloud.storage.Conversions.Decoder;
+import com.google.cloud.storage.Crc32cValue.Crc32cLengthKnown;
+import com.google.cloud.storage.Retrying.RetryingDependencies;
+import com.google.cloud.storage.UnbufferedWritableByteChannelSession.UnbufferedWritableByteChannel;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import com.google.protobuf.ByteString;
+import com.google.storage.v2.ChecksummedData;
+import com.google.storage.v2.Object;
+import com.google.storage.v2.ObjectChecksums;
+import com.google.storage.v2.QueryWriteStatusRequest;
+import com.google.storage.v2.QueryWriteStatusResponse;
+import com.google.storage.v2.WriteObjectRequest;
+import com.google.storage.v2.WriteObjectResponse;
+import io.grpc.Status.Code;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.ClosedChannelException;
+import java.nio.channels.GatheringByteChannel;
+import java.nio.channels.SeekableByteChannel;
+import java.util.Arrays;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.checkerframework.checker.nullness.qual.NonNull;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+final class SyncAndUploadUnbufferedWritableByteChannel implements UnbufferedWritableByteChannel {
+
+  private final ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write;
+  private final UnaryCallable<QueryWriteStatusRequest, QueryWriteStatusResponse> query;
+  private final SettableApiFuture<WriteObjectResponse> resultFuture;
+  private final ChunkSegmenter chunkSegmenter;
+  private final WriteCtx<ResumableWrite> writeCtx;
+  private final RetryingDependencies deps;
+  private final ResultRetryAlgorithm<?> alg;
+  private final RecoveryFile rf;
+
+  private final String uploadId;
+  private final BufferHandle copyBuffer;
+
+  /* --- running state --- */
+  private final RequestStream stream;
+
+  private boolean open;
+  private @Nullable GatheringByteChannel sync;
+  private boolean first;
+  private boolean finished;
+
+  SyncAndUploadUnbufferedWritableByteChannel(
+      ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write,
+      UnaryCallable<QueryWriteStatusRequest, QueryWriteStatusResponse> query,
+      SettableApiFuture<WriteObjectResponse> resultFuture,
+      ChunkSegmenter chunkSegmenter,
+      RetryingDependencies deps,
+      ResultRetryAlgorithm<?> alg,
+      WriteCtx<ResumableWrite> writeCtx,
+      RecoveryFile rf,
+      BufferHandle copyBuffer) {
+    this.write =
+        write.withDefaultCallContext(
+            contextWithBucketName(
+                writeCtx.getRequestFactory().bucketName(), GrpcCallContext.createDefault()));
+    this.query = query;
+    this.resultFuture = resultFuture;
+    this.chunkSegmenter = chunkSegmenter;
+    this.writeCtx = writeCtx;
+    this.deps = deps;
+    this.alg = new Alg(alg, resultFuture);
+    this.rf = rf;
+    this.uploadId = writeCtx.newRequestBuilder().getUploadId();
+    this.copyBuffer = copyBuffer;
+    this.stream = new RequestStream(this.write, resultFuture);
+    this.open = true;
+    this.first = true;
+    this.finished = false;
+  }
+
+  @SuppressWarnings("resource")
+  @Override
+  public long write(ByteBuffer[] srcs, int offset, int length) throws IOException {
+    if (!open) {
+      throw new ClosedChannelException();
+    }
+    ByteBuffer[] duplicates =
+        Arrays.stream(srcs, offset, offset + length)
+            .map(ByteBuffer::duplicate)
+            .toArray(ByteBuffer[]::new);
+    long prevWritten = writeCtx.getTotalSentBytes().get();
+    long syncWritten = openSync().write(duplicates);
+    long goalSize = Math.addExact(prevWritten, syncWritten);
+    ChunkSegment[] segments = chunkSegmenter.segmentBuffers(srcs, offset, length);
+    doUpload(false, segments, goalSize);
+    return syncWritten;
+  }
+
+  @Override
+  public boolean isOpen() {
+    return open;
+  }
+
+  @Override
+  public void close() throws IOException {
+    if (!open) {
+      return;
+    }
+    try {
+      doUpload(true, new ChunkSegment[0], writeCtx.getTotalSentBytes().get());
+      rf.close();
+    } finally {
+      open = false;
+    }
+  }
+
+  private GatheringByteChannel openSync() throws IOException {
+    if (sync == null) {
+      sync = rf.syncingChannel();
+    }
+    return sync;
+  }
+
+  private WriteObjectRequest processSegment(ChunkSegment segment) {
+    WriteObjectRequest.Builder builder = writeCtx.newRequestBuilder();
+    if (!first) {
+      builder.clearUploadId().clearWriteObjectSpec().clearObjectChecksums();
+    } else {
+      first = false;
+    }
+
+    Crc32cLengthKnown crc32c = segment.getCrc32c();
+    ByteString b = segment.getB();
+    int contentSize = b.size();
+
+    // update ctx state that tracks overall progress
+    writeCtx
+        .getCumulativeCrc32c()
+        .accumulateAndGet(crc32c, chunkSegmenter.getHasher()::nullSafeConcat);
+    // resolve current offset and set next
+    long offset = writeCtx.getTotalSentBytes().getAndAdd(contentSize);
+
+    ChecksummedData.Builder checksummedData = ChecksummedData.newBuilder().setContent(b);
+    if (crc32c != null) {
+      checksummedData.setCrc32C(crc32c.getValue());
+    }
+    builder.setWriteOffset(offset).setChecksummedData(checksummedData.build());
+
+    if (!segment.isOnlyFullBlocks()) {
+      finishMessage(builder);
+      finished = true;
+    }
+
+    WriteObjectRequest build = builder.build();
+    return build;
+  }
+
+  @NonNull
+  private WriteObjectRequest finishMessage() {
+    long offset = writeCtx.getTotalSentBytes().get();
+
+    WriteObjectRequest.Builder b = writeCtx.newRequestBuilder().setWriteOffset(offset);
+
+    WriteObjectRequest message = finishMessage(b).build();
+    return message;
+  }
+
+  private WriteObjectRequest.Builder finishMessage(WriteObjectRequest.Builder b) {
+    Crc32cLengthKnown crc32cValue = writeCtx.getCumulativeCrc32c().get();
+    b.setFinishWrite(true);
+    if (crc32cValue != null) {
+      b.setObjectChecksums(ObjectChecksums.newBuilder().setCrc32C(crc32cValue.getValue()).build());
+    }
+    return b;
+  }
+
+  private void doUpload(boolean closing, ChunkSegment[] segments, long goalSize) {
+    AtomicBoolean recover = new AtomicBoolean(false);
+    Retrying.run(
+        deps,
+        alg,
+        () -> {
+          if (closing && sync != null) {
+            sync.close();
+          }
+          boolean shouldRecover = recover.getAndSet(true);
+          if (!shouldRecover) {
+            for (ChunkSegment segment : segments) {
+              WriteObjectRequest writeObjectRequest = processSegment(segment);
+              stream.onNext(writeObjectRequest);
+            }
+
+            if (closing && !finished) {
+              WriteObjectRequest message = finishMessage();
+              stream.onNext(message);
+              finished = true;
+            }
+
+            if (closing) {
+              stream.onCompleted();
+            }
+          } else {
+            if (sync != null) {
+              sync.close();
+              sync = null;
+            }
+            stream.reset();
+
+            QueryWriteStatusRequest req =
+                QueryWriteStatusRequest.newBuilder().setUploadId(uploadId).build();
+            QueryWriteStatusResponse resp = query.call(req);
+            // if the response has a resource the session completed, no need to re-upload
+            if (!resp.hasResource()) {
+              long persistedSize = resp.getPersistedSize();
+
+              if (persistedSize != goalSize) {
+
+                // rewind our context
+                finished = false;
+                first = true;
+                writeCtx.getTotalSentBytes().set(persistedSize);
+                writeCtx.getConfirmedBytes().set(persistedSize);
+                writeCtx.getCumulativeCrc32c().set(null); // todo: can we rewind checksum?
+
+                try (SeekableByteChannel reader = rf.reader()) {
+                  reader.position(persistedSize);
+                  ByteBuffer buf = copyBuffer.get();
+                  while (Buffers.fillFrom(buf, reader) != -1) {
+                    buf.flip();
+                    while (buf.hasRemaining()) {
+                      ChunkSegment[] recoverySegments = chunkSegmenter.segmentBuffer(buf);
+                      for (ChunkSegment segment : recoverySegments) {
+                        WriteObjectRequest writeObjectRequest = processSegment(segment);
+                        stream.onNext(writeObjectRequest);
+                      }
+                    }
+                    buf.clear();
+                  }
+                }
+                if (closing && !finished) {
+                  WriteObjectRequest message = finishMessage();
+                  stream.onNext(message);
+                  finished = true;
+                }
+                recover.compareAndSet(true, false);
+                if (closing || finished) {
+                  stream.onCompleted();
+                }
+              }
+            } else {
+              Object resource = resp.getResource();
+              resultFuture.set(WriteObjectResponse.newBuilder().setResource(resource).build());
+            }
+          }
+          long newWritten = writeCtx.getTotalSentBytes().get();
+          Preconditions.checkState(newWritten == goalSize, "%s == %s", newWritten, goalSize);
+          return null;
+        },
+        Decoder.identity());
+  }
+
+  @VisibleForTesting
+  static final class RequestStream implements ApiStreamObserver<WriteObjectRequest> {
+    private static final ApiException CLIENT_RESET_ERROR =
+        ApiExceptionFactory.createException(null, GrpcStatusCode.of(Code.ABORTED), false);
+
+    private final ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write;
+    private final SettableApiFuture<WriteObjectResponse> resultFuture;
+
+    private volatile StreamPair streamPair;
+
+    private RequestStream(
+        ClientStreamingCallable<WriteObjectRequest, WriteObjectResponse> write,
+        SettableApiFuture<WriteObjectResponse> resultFuture) {
+      this.write = write;
+      this.resultFuture = resultFuture;
+    }
+
+    @Override
+    public void onNext(WriteObjectRequest value) {
+      StreamPair pair = ensureOpen();
+      Throwable err = pair.getResponseStream().error;
+      if (err != null) {
+        reset();
+        throw StorageException.coalesce(err);
+      }
+
+      halfClosedToUnavailable(() -> pair.getRequestStream().onNext(value));
+    }
+
+    @Override
+    public void onError(Throwable t) {
+      try {
+        halfClosedToUnavailable(() -> ensureOpen().getRequestStream().onError(t));
+      } finally {
+        streamPair = null;
+      }
+    }
+
+    @Override
+    public void onCompleted() {
+      StreamPair pair = ensureOpen();
+      Throwable err = pair.getResponseStream().error;
+      if (err != null) {
+        reset();
+        throw StorageException.coalesce(err);
+      }
+
+      halfClosedToUnavailable(pair.getRequestStream()::onCompleted);
+      pair.getResponseStream().await();
+    }
+
+    private StreamPair ensureOpen() {
+      if (streamPair == null) {
+        ResponseStream responseStream = new ResponseStream(resultFuture);
+        ApiStreamObserver<WriteObjectRequest> requestStream =
+            write.clientStreamingCall(responseStream);
+        streamPair = new StreamPair(requestStream, responseStream);
+      }
+      return streamPair;
+    }
+
+    private void reset() {
+      if (streamPair != null && streamPair.getRequestStream() != null) {
+        streamPair.getRequestStream().onError(CLIENT_RESET_ERROR);
+      }
+      streamPair = null;
+    }
+
+    /**
+     * If the stream is in the process of closing (usually due to error) and we call a method on it
+     * we will receive an IllegalStateException. A stream being half closed is not a terminal state
+     * for our upload operation. Attempt to detect and translate it into an UNAVAILABLE error we can
+     * retry.
+     */
+    static void halfClosedToUnavailable(Runnable r) {
+      try {
+        r.run();
+      } catch (IllegalStateException ise) {
+        String message = ise.getMessage();
+        if (message != null && message.contains("half-closed")) {
+          throw ApiExceptionFactory.createException(ise, GrpcStatusCode.of(Code.UNAVAILABLE), true);
+        } else {
+          throw ise;
+        }
+      }
+    }
+  }
+
+  @VisibleForTesting
+  static final class ResponseStream implements ApiStreamObserver<WriteObjectResponse> {
+    private final SettableApiFuture<Void> invocationHandle;
+    private final SettableApiFuture<WriteObjectResponse> resultFuture;
+
+    private volatile WriteObjectResponse last;
+    private volatile Throwable error;
+
+    @VisibleForTesting
+    ResponseStream(SettableApiFuture<WriteObjectResponse> resultFuture) {
+      this.resultFuture = resultFuture;
+      this.invocationHandle = SettableApiFuture.create();
+    }
+
+    @Override
+    public void onNext(WriteObjectResponse value) {
+      last = value;
+      error = null;
+    }
+
+    @Override
+    public void onError(Throwable t) {
+      error = t;
+      invocationHandle.setException(t);
+    }
+
+    @Override
+    public void onCompleted() {
+      if (last != null && last.hasResource()) {
+        resultFuture.set(last);
+      }
+      invocationHandle.set(null);
+    }
+
+    void await() {
+      ApiExceptions.callAndTranslateApiException(invocationHandle);
+    }
+  }
+
+  @VisibleForTesting
+  static final class Alg implements ResultRetryAlgorithm<WriteObjectResponse> {
+
+    private final ResultRetryAlgorithm<WriteObjectResponse> delegate;
+    private final SettableApiFuture<WriteObjectResponse> resultFuture;
+
+    @VisibleForTesting
+    @SuppressWarnings("unchecked")
+    Alg(ResultRetryAlgorithm<?> delegate, SettableApiFuture<WriteObjectResponse> resultFuture) {
+      this.delegate = (ResultRetryAlgorithm<WriteObjectResponse>) delegate;
+      this.resultFuture = resultFuture;
+    }
+
+    @Override
+    public TimedAttemptSettings createNextAttempt(
+        Throwable prevThrowable,
+        WriteObjectResponse prevResponse,
+        TimedAttemptSettings prevSettings) {
+      return delegate.createNextAttempt(prevThrowable, prevResponse, prevSettings);
+    }
+
+    @Override
+    public boolean shouldRetry(Throwable prevThrowable, WriteObjectResponse prevResponse)
+        throws CancellationException {
+      boolean shouldRetry = delegate.shouldRetry(prevThrowable, prevResponse);
+      if (!shouldRetry && prevThrowable != null) {
+        resultFuture.setException(prevThrowable);
+      }
+      return shouldRetry;
+    }
+  }
+
+  private static final class StreamPair {
+    private final ApiStreamObserver<WriteObjectRequest> requestStream;
+    private final ResponseStream responseStream;
+
+    private StreamPair(
+        ApiStreamObserver<WriteObjectRequest> requestStream, ResponseStream responseStream) {
+      this.requestStream = requestStream;
+      this.responseStream = responseStream;
+    }
+
+    public ApiStreamObserver<WriteObjectRequest> getRequestStream() {
+      return requestStream;
+    }
+
+    public ResponseStream getResponseStream() {
+      return responseStream;
+    }
+  }
+}
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
index 776629cf..bf9d523b 100644
--- 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
@@ -18,11 +18,14 @@ package com.google.cloud.storage;
 
 import com.google.common.base.MoreObjects;
 import java.io.IOException;
+import java.nio.Buffer;
 import java.nio.ByteBuffer;
+import java.nio.channels.GatheringByteChannel;
 import java.nio.channels.WritableByteChannel;
 import java.time.Clock;
 import java.time.Duration;
 import java.time.Instant;
+import java.util.Arrays;
 import java.util.Objects;
 import java.util.logging.Logger;
 
@@ -36,6 +39,8 @@ interface ThroughputSink {
 
   WritableByteChannel decorate(WritableByteChannel wbc);
 
+  GatheringByteChannel decorate(GatheringByteChannel gbc);
+
   static void computeThroughput(Clock clock, ThroughputSink sink, long numBytes, IO io)
       throws IOException {
     boolean exception = false;
@@ -171,6 +176,11 @@ interface ThroughputSink {
       return new ThroughputRecordingWritableByteChannel(wbc, this, clock);
     }
 
+    @Override
+    public GatheringByteChannel decorate(GatheringByteChannel gbc) {
+      return new ThroughputRecordingGatheringByteChannel(gbc, this, clock);
+    }
+
     @Override
     public String toString() {
       return MoreObjects.toStringHelper(this).add("prefix", prefix).add("clock", clock).toString();
@@ -191,6 +201,30 @@ interface ThroughputSink {
 
     @Override
     public int write(ByteBuffer src) throws IOException {
+      return ThroughputRecordingWritableByteChannel.write(src, clock, delegate, sink);
+    }
+
+    @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();
+    }
+
+    static int write(ByteBuffer src, Clock clock, WritableByteChannel delegate, ThroughputSink sink)
+        throws IOException {
       boolean exception = false;
       int remaining = src.remaining();
       Instant begin = clock.instant();
@@ -205,6 +239,60 @@ interface ThroughputSink {
         sink.recordThroughput(record);
       }
     }
+  }
+
+  final class ThroughputRecordingGatheringByteChannel implements GatheringByteChannel {
+    private final GatheringByteChannel delegate;
+    private final ThroughputSink sink;
+    private final Clock clock;
+
+    private ThroughputRecordingGatheringByteChannel(
+        GatheringByteChannel delegate, ThroughputSink sink, Clock clock) {
+      this.delegate = delegate;
+      this.sink = sink;
+      this.clock = clock;
+    }
+
+    @Override
+    public int write(ByteBuffer src) throws IOException {
+      return ThroughputRecordingWritableByteChannel.write(src, clock, delegate, sink);
+    }
+
+    @Override
+    public long write(ByteBuffer[] srcs, int offset, int length) throws IOException {
+      boolean exception = false;
+      long available = Arrays.stream(srcs).mapToLong(Buffer::remaining).sum();
+      Instant begin = clock.instant();
+      try {
+        return delegate.write(srcs, offset, length);
+      } catch (IOException e) {
+        exception = true;
+        throw e;
+      } finally {
+        Instant end = clock.instant();
+        long remaining = Arrays.stream(srcs).mapToLong(Buffer::remaining).sum();
+        Record record = Record.of(available - remaining, begin, end, exception);
+        sink.recordThroughput(record);
+      }
+    }
+
+    @Override
+    public long write(ByteBuffer[] srcs) throws IOException {
+      boolean exception = false;
+      long available = Arrays.stream(srcs).mapToLong(Buffer::remaining).sum();
+      Instant begin = clock.instant();
+      try {
+        return delegate.write(srcs);
+      } catch (IOException e) {
+        exception = true;
+        throw e;
+      } finally {
+        Instant end = clock.instant();
+        long remaining = Arrays.stream(srcs).mapToLong(Buffer::remaining).sum();
+        Record record = Record.of(available - remaining, begin, end, exception);
+        sink.recordThroughput(record);
+      }
+    }
 
     @Override
     public boolean isOpen() {
@@ -246,6 +334,11 @@ interface ThroughputSink {
       return b.decorate(a.decorate(wbc));
     }
 
+    @Override
+    public GatheringByteChannel decorate(GatheringByteChannel gbc) {
+      return b.decorate(a.decorate(gbc));
+    }
+
     @Override
     public String toString() {
       return MoreObjects.toStringHelper(this).add("a", a).add("b", b).toString();
@@ -271,6 +364,11 @@ interface ThroughputSink {
       return new ThroughputRecordingWritableByteChannel(wbc, this, clock);
     }
 
+    @Override
+    public GatheringByteChannel decorate(GatheringByteChannel gbc) {
+      return new ThroughputRecordingGatheringByteChannel(gbc, this, clock);
+    }
+
     @Override
     public String toString() {
       return MoreObjects.toStringHelper(this).add("w", w).add("clock", clock).toString();
@@ -289,5 +387,10 @@ interface ThroughputSink {
     public WritableByteChannel decorate(WritableByteChannel wbc) {
       return wbc;
     }
+
+    @Override
+    public GatheringByteChannel decorate(GatheringByteChannel gbc) {
+      return gbc;
+    }
   }
 }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/WriteCtx.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/WriteCtx.java
index 2488cfcb..654811f4 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/WriteCtx.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/WriteCtx.java
@@ -40,6 +40,10 @@ final class WriteCtx<RequestFactoryT extends WriteObjectRequestBuilderFactory> {
     this.cumulativeCrc32c = new AtomicReference<>();
   }
 
+  public RequestFactoryT getRequestFactory() {
+    return requestFactory;
+  }
+
   public WriteObjectRequest.Builder newRequestBuilder() {
     return requestFactory.newBuilder();
   }
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/WriteFlushStrategy.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/WriteFlushStrategy.java
index b77128d8..806c2b42 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/WriteFlushStrategy.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/WriteFlushStrategy.java
@@ -86,8 +86,7 @@ final class WriteFlushStrategy {
         new FsyncOnClose(write, bucketName, committedTotalBytesCallback, onSuccessCallback);
   }
 
-  private static GrpcCallContext contextWithBucketName(
-      String bucketName, GrpcCallContext baseContext) {
+  static GrpcCallContext contextWithBucketName(String bucketName, GrpcCallContext baseContext) {
     if (bucketName != null && !bucketName.isEmpty()) {
       return baseContext.withExtraHeaders(
           ImmutableMap.of(
@@ -257,7 +256,7 @@ final class WriteFlushStrategy {
     }
   }
 
-  private static class Observer implements ApiStreamObserver<WriteObjectResponse> {
+  static class Observer implements ApiStreamObserver<WriteObjectResponse> {
 
     private final LongConsumer sizeCallback;
     private final Consumer<WriteObjectResponse> completeCallback;
@@ -265,7 +264,7 @@ final class WriteFlushStrategy {
     private final SettableApiFuture<Void> invocationHandle;
     private volatile WriteObjectResponse last;
 
-    private Observer(LongConsumer sizeCallback, Consumer<WriteObjectResponse> completeCallback) {
+    Observer(LongConsumer sizeCallback, Consumer<WriteObjectResponse> completeCallback) {
       this.sizeCallback = sizeCallback;
       this.completeCallback = completeCallback;
       this.invocationHandle = SettableApiFuture.create();
@@ -304,7 +303,7 @@ final class WriteFlushStrategy {
       invocationHandle.set(null);
     }
 
-    private void await() {
+    void await() {
       try {
         invocationHandle.get();
       } catch (InterruptedException | ExecutionException e) {
diff --git a/pom.xml b/pom.xml
index 409694dc..029eb17d 100644
--- a/pom.xml
+++ b/pom.xml
@@ -54,7 +54,7 @@
     <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
     <github.global.server>github</github.global.server>
     <site.installationModule>google-cloud-storage-parent</site.installationModule>
-    <google.cloud.shared-dependencies.version>3.14.0</google.cloud.shared-dependencies.version>
+    <google.cloud.shared-dependencies.version>3.15.0</google.cloud.shared-dependencies.version>
   </properties>
 
   <dependencyManagement>
@@ -81,7 +81,7 @@
       <dependency>
         <groupId>com.google.apis</groupId>
         <artifactId>google-api-services-storage</artifactId>
-        <version>v1-rev20230710-2.0.0</version>
+        <version>v1-rev20230907-2.0.0</version>
       </dependency>
       <dependency>
         <groupId>com.google.cloud</groupId>
diff --git a/proto-google-cloud-storage-v2/src/main/java/com/google/storage/v2/BucketName.java b/proto-google-cloud-storage-v2/src/main/java/com/google/storage/v2/BucketName.java
index 078819ae..d2643c96 100644
--- a/proto-google-cloud-storage-v2/src/main/java/com/google/storage/v2/BucketName.java
+++ b/proto-google-cloud-storage-v2/src/main/java/com/google/storage/v2/BucketName.java
@@ -137,7 +137,7 @@ public class BucketName implements ResourceName {
     if (o == this) {
       return true;
     }
-    if (o != null || getClass() == o.getClass()) {
+    if (o != null && getClass() == o.getClass()) {
       BucketName that = ((BucketName) o);
       return Objects.equals(this.project, that.project) && Objects.equals(this.bucket, that.bucket);
     }
diff --git a/proto-google-cloud-storage-v2/src/main/java/com/google/storage/v2/CryptoKeyName.java b/proto-google-cloud-storage-v2/src/main/java/com/google/storage/v2/CryptoKeyName.java
index f80912dc..a9d2ffd7 100644
--- a/proto-google-cloud-storage-v2/src/main/java/com/google/storage/v2/CryptoKeyName.java
+++ b/proto-google-cloud-storage-v2/src/main/java/com/google/storage/v2/CryptoKeyName.java
@@ -175,7 +175,7 @@ public class CryptoKeyName implements ResourceName {
     if (o == this) {
       return true;
     }
-    if (o != null || getClass() == o.getClass()) {
+    if (o != null && getClass() == o.getClass()) {
       CryptoKeyName that = ((CryptoKeyName) o);
       return Objects.equals(this.project, that.project)
           && Objects.equals(this.location, that.location)
diff --git a/proto-google-cloud-storage-v2/src/main/java/com/google/storage/v2/NotificationConfigName.java b/proto-google-cloud-storage-v2/src/main/java/com/google/storage/v2/NotificationConfigName.java
index ecb00bec..3a0c0a36 100644
--- a/proto-google-cloud-storage-v2/src/main/java/com/google/storage/v2/NotificationConfigName.java
+++ b/proto-google-cloud-storage-v2/src/main/java/com/google/storage/v2/NotificationConfigName.java
@@ -159,7 +159,7 @@ public class NotificationConfigName implements ResourceName {
     if (o == this) {
       return true;
     }
-    if (o != null || getClass() == o.getClass()) {
+    if (o != null && getClass() == o.getClass()) {
       NotificationConfigName that = ((NotificationConfigName) o);
       return Objects.equals(this.project, that.project)
           && Objects.equals(this.bucket, that.bucket)
diff --git a/proto-google-cloud-storage-v2/src/main/java/com/google/storage/v2/ProjectName.java b/proto-google-cloud-storage-v2/src/main/java/com/google/storage/v2/ProjectName.java
index 154bd94b..b8d7aa7b 100644
--- a/proto-google-cloud-storage-v2/src/main/java/com/google/storage/v2/ProjectName.java
+++ b/proto-google-cloud-storage-v2/src/main/java/com/google/storage/v2/ProjectName.java
@@ -127,7 +127,7 @@ public class ProjectName implements ResourceName {
     if (o == this) {
       return true;
     }
-    if (o != null || getClass() == o.getClass()) {
+    if (o != null && getClass() == o.getClass()) {
       ProjectName that = ((ProjectName) o);
       return Objects.equals(this.project, that.project);
     }


Test patch:
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/TestUtils.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/TestUtils.java
index af92080a..fb14f711 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/TestUtils.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/TestUtils.java
@@ -229,6 +229,10 @@ public final class TestUtils {
     return xxd(true, bytes);
   }
 
+  public static String xxd(ByteString bytes) {
+    return xxd(false, bytes.asReadOnlyByteBuffer());
+  }
+
   public static String xxd(boolean flip, ByteBuffer bytes) {
     ByteBuffer dup = bytes.duplicate();
     if (flip) dup.flip();
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
index a7840f45..a0074006 100644
--- 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
@@ -22,6 +22,7 @@ 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.GatheringByteChannel;
 import java.nio.channels.WritableByteChannel;
 import java.time.Duration;
 import java.time.Instant;
@@ -258,5 +259,10 @@ public final class ThroughputSinkTest {
     public WritableByteChannel decorate(WritableByteChannel wbc) {
       return null;
     }
+
+    @Override
+    public GatheringByteChannel decorate(GatheringByteChannel wbc) {
+      return null;
+    }
   }
 }
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITGrpcInterceptorTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITGrpcInterceptorTest.java
index f7c9d1f1..737d31dd 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITGrpcInterceptorTest.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITGrpcInterceptorTest.java
@@ -83,8 +83,6 @@ public class ITGrpcInterceptorTest {
             .map(m -> m.get(X_GOOG_REQUEST_PARAMS))
             .collect(Collectors.toList());
 
-    System.out.println("requestParams = " + requestParams);
-
     String expected = String.format("project=projects/%s", options.getProjectId());
     String expectedEncoded = String.format("project=projects%%2F%s", options.getProjectId());
     assertThat(requestParams).containsAnyOf(expected, expectedEncoded);
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITJournalingBlobWriteSessionConfigTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITJournalingBlobWriteSessionConfigTest.java
new file mode 100644
index 00000000..4526a09f
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITJournalingBlobWriteSessionConfigTest.java
@@ -0,0 +1,129 @@
+/*
+ * 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.it;
+
+import static com.google.cloud.storage.TestUtils.assertAll;
+import static com.google.common.truth.Truth.assertThat;
+
+import com.google.cloud.storage.BlobInfo;
+import com.google.cloud.storage.BlobWriteSession;
+import com.google.cloud.storage.BlobWriteSessionConfigs;
+import com.google.cloud.storage.BucketInfo;
+import com.google.cloud.storage.DataGenerator;
+import com.google.cloud.storage.GrpcStorageOptions;
+import com.google.cloud.storage.JournalingBlobWriteSessionConfig;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.Storage.BlobWriteOption;
+import com.google.cloud.storage.TransportCompatibility.Transport;
+import com.google.cloud.storage.it.runner.StorageITRunner;
+import com.google.cloud.storage.it.runner.annotations.Backend;
+import com.google.cloud.storage.it.runner.annotations.Inject;
+import com.google.cloud.storage.it.runner.annotations.SingleBackend;
+import com.google.cloud.storage.it.runner.annotations.StorageFixture;
+import com.google.cloud.storage.it.runner.registry.Generator;
+import com.google.common.collect.ImmutableList;
+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.util.Random;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Stream;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+
+@RunWith(StorageITRunner.class)
+@SingleBackend(Backend.PROD)
+public final class ITJournalingBlobWriteSessionConfigTest {
+  private static final int _1MiB = 1024 * 1024;
+  private static final int _256MiB = 256 * _1MiB;
+
+  @Inject
+  @StorageFixture(Transport.GRPC)
+  public Storage storage;
+
+  @Inject public BucketInfo bucket;
+  @Inject public Generator generator;
+
+  @Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+  private Storage journalingStorage;
+  private Path tempDir;
+
+  @Before
+  public void setUp() throws Exception {
+    tempDir = temporaryFolder.newFolder(generator.randomObjectName()).toPath();
+    JournalingBlobWriteSessionConfig journaling =
+        BlobWriteSessionConfigs.journaling(ImmutableList.of(tempDir));
+    journalingStorage =
+        ((GrpcStorageOptions.Builder) this.storage.getOptions().toBuilder())
+            .setBlobWriteSessionConfig(journaling)
+            .build()
+            .getService();
+  }
+
+  @After
+  public void tearDown() throws Exception {
+    if (journalingStorage != null) {
+      journalingStorage.close();
+    }
+  }
+
+  @Test
+  public void allBytesProperlyTransmitted() throws Exception {
+
+    Random rand = new Random(bucket.getName().hashCode());
+    int objectSize = _256MiB;
+    byte[] bytes = DataGenerator.rand(rand).genBytes(objectSize);
+
+    BlobInfo info = BlobInfo.newBuilder(bucket, generator.randomObjectName()).build();
+
+    BlobWriteSession blobWriteSession =
+        journalingStorage.blobWriteSession(info, BlobWriteOption.doesNotExist());
+    try (WritableByteChannel w = blobWriteSession.open()) {
+      w.write(ByteBuffer.wrap(bytes));
+    }
+
+    BlobInfo resource = blobWriteSession.getResult().get(1, TimeUnit.SECONDS);
+
+    byte[] actual = storage.readAllBytes(info.getBlobId());
+    assertAll(
+        () -> assertThat(resource.getSize()).isEqualTo(objectSize),
+        () -> assertThat(actual).isEqualTo(bytes));
+  }
+
+  @Test
+  public void journalFileMustNotBeLeftOnDiskAfterSuccess() throws IOException {
+    BlobInfo info = BlobInfo.newBuilder(bucket, generator.randomObjectName()).build();
+    byte[] bytes = DataGenerator.base64Characters().genBytes(17);
+    BlobWriteSession blobWriteSession =
+        journalingStorage.blobWriteSession(info, BlobWriteOption.doesNotExist());
+    try (WritableByteChannel w = blobWriteSession.open()) {
+      w.write(ByteBuffer.wrap(bytes));
+    }
+
+    try (Stream<Path> stream = Files.list(tempDir)) {
+      ImmutableList<Path> leftOverFiles = stream.collect(ImmutableList.toImmutableList());
+      assertThat(leftOverFiles).isEmpty();
+    }
+  }
+}