whowes · Oct 16, 2026
@@ -359,6 +359,66 @@ void testProgressListener(@TempDir Path tempDir) throws Exception {
     }
   }
 
+  @Test
+  void testResumeUpload(@TempDir Path tempDir) throws Exception {
+    int totalBytes = 600 * 1024;
+    int unalignedChunkSize = 300 * 1024;
+    Path file = createTempFile(tempDir, "it-resume.txt", totalBytes);
+    UploadMediaRequest request = UploadMediaRequest.newBuilder().setName("it-resume.txt").build();
+    ApiCallContext failSecondChunkContext = faultInjectionContext(403, 1, SHOWCASE_CHUNK_SIZE);
+
+    ResumableUploadFuture<UploadMediaResponse> firstAttemptFuture =
+        client
+            .uploadMediaCallable()
+            .futureCall(
+                request,
+                () -> Files.newInputStream(file),
+                failSecondChunkContext,
+                ResumableUploadOptions.newBuilder().setChunkSize(unalignedChunkSize).build());
+
+    assertThrows(ExecutionException.class, () -> firstAttemptFuture.get(15, TimeUnit.SECONDS));
+    String sessionUrl = firstAttemptFuture.getUploadSessionUrl();
+    Integer effectiveChunkSize = firstAttemptFuture.getEffectiveChunkSize();
+    assertThat(sessionUrl).isNotNull();
+    assertThat(effectiveChunkSize).isEqualTo(SHOWCASE_CHUNK_SIZE);
+
+    List<ResumableUploadProgress> resumeProgress = new CopyOnWriteArrayList<>();
+    CountDownLatch finalizedLatch = new CountDownLatch(1);
+    ResumableUploadOptions resumeOptions =
+        ResumableUploadOptions.newBuilder()
+            .setChunkSize(effectiveChunkSize)
+            .setProgressListener(
+                status -> {
+                  resumeProgress.add(status);
+                  if ("FINALIZED".equals(status.getState())) {
+                    finalizedLatch.countDown();
+                  }
+                })
+            .build();
+
+    ResumableUploadFuture<UploadMediaResponse> secondAttemptFuture =
+        client
+            .uploadMediaCallable()
+            .resumeCall(sessionUrl, () -> Files.newInputStream(file), resumeOptions);
+    UploadMediaResponse response = secondAttemptFuture.get(30, TimeUnit.SECONDS);
+    assertThat(finalizedLatch.await(10, TimeUnit.SECONDS)).isTrue();
+
+    assertThat(response.getName()).isEqualTo("it-resume.txt");
+    assertThat(response.getSize()).isEqualTo(totalBytes);
+    assertThat(
+            resumeProgress.stream()
+                .map(ResumableUploadProgress::getState)
+                .collect(Collectors.toList()))
+        .containsExactly("RECOVERING", "OFFSET_RECEIVED", "UPLOADING", "FINALIZED")
+        .inOrder();
+    assertThat(
+            resumeProgress.stream()
+                .map(ResumableUploadProgress::getBytesUploaded)
+                .collect(Collectors.toList()))
+        .containsExactly(0L, 262144L, 524288L, 614400L)
+        .inOrder();
+  }
+
   private static Path createTempFile(Path dir, String fileName, byte[] data) throws IOException {
     Path path = dir.resolve(fileName);
     Files.write(path, data);

@@ -228,6 +228,7 @@ public void onClose(int statusCode, HttpJsonMetadata trailers) {
             future.set(
                 QueryStatusResponse.<ResponseT>newBuilder()
                     .setUploadStatus(uploadStatus)
+                    .setCommittedOffset(committedOffset)
                     .setResponse(responseParser.parse(stream))
                     .build());
           } else if (committedOffset != null) {

@@ -43,13 +43,14 @@
 public interface InputStreamSupplier {
 
   /**
-   * Opens or returns the {@link InputStream} to be read.
+   * Opens or returns the {@link InputStream} to be read, positioned at the beginning of the
+   * payload.
    *
    * <p>The returned stream is managed by the resumable upload and is closed automatically when the
    * upload succeeds, fails, times out, or is cancelled. Callers should not close the stream
    * themselves.
    *
-   * @return a non-null input stream over the payload data
+   * @return a non-null input stream positioned at byte {@code 0} of the payload data
    * @throws IOException if an I/O error occurs when opening or obtaining the stream
    */
   InputStream get() throws IOException;

@@ -85,6 +85,16 @@ public abstract ResumableUploadFuture<ResponseT> futureCall(
   /**
    * Resumes an existing resumable upload session asynchronously using a saved session URL.
    *
+   * <p>Resuming queries the server for the last committed byte offset and skips the stream forward
+   * to that offset before resuming chunk transmission. {@code payload} must supply a stream
+   * positioned at byte {@code 0} of the original payload.
+   *
+   * <p>Because server chunk granularity is not re-derived on resume, callers resuming an upload
+   * should configure the initial session's effective chunk size (from {@link
+   * ResumableUploadFuture#getEffectiveChunkSize()} or {@link
+   * ResumableUploadProgress#getEffectiveChunkSize()}) via {@link
+   * ResumableUploadOptions.Builder#setChunkSize(int)}.
+   *
    * <p>Streams from the provided {@code payload} are consumed asynchronously by the returned {@link
    * ResumableUploadFuture} and will be closed automatically upon completion, failure, or
    * cancellation.

@@ -29,6 +29,7 @@
  */
 package com.google.api.gax.rpc;
 
+import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Preconditions.checkNotNull;
 
 import com.google.api.core.ApiFuture;
@@ -45,6 +46,7 @@
 import com.google.api.gax.retrying.RetryAlgorithm;
 import com.google.api.gax.retrying.RetrySettings;
 import com.google.api.gax.retrying.ScheduledRetryingExecutor;
+import com.google.common.base.Strings;
 import java.time.Duration;
 import org.jspecify.annotations.NullMarked;
 import org.jspecify.annotations.Nullable;
@@ -131,7 +133,18 @@ public ResumableUploadFuture<ResponseT> futureCall(
   @Override
   public ResumableUploadFuture<ResponseT> resumeCall(
       String sessionUrl, InputStreamSupplier payload, @Nullable ResumableUploadOptions options) {
-    throw new UnsupportedOperationException("Session resumption is not yet implemented.");
+    checkArgument(!Strings.isNullOrEmpty(sessionUrl), "sessionUrl must not be null or empty");
+    checkNotNull(payload, "payload must not be null");
+    ResumableUploadOptions effectiveOptions = defaultOptions.merge(options);
+
+    return ResumableUploadFutureImpl.createForResume(
+        sessionUrl,
+        retryingUploadChunkCallable,
+        retryingQueryCallable,
+        payload,
+        effectiveOptions,
+        clientContext,
+        recoveryAlgorithm);
   }
 
   private static <RequestT, ResponseT> UnaryCallable<RequestT, ResponseT> createRetryingCallable(

@@ -47,6 +47,7 @@
 import com.google.common.util.concurrent.MoreExecutors;
 import java.io.IOException;
 import java.io.InputStream;
+import java.util.Optional;
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.Future;
 import java.util.concurrent.ScheduledExecutorService;
@@ -119,6 +120,11 @@ void start() {
     }
   }
 
+  void resume() {
+    registerCancellationListener();
+    queryStatus();
+  }
+
   private void registerCancellationListener() {
     uploadResultFuture.addListener(
         () -> {
@@ -271,7 +277,10 @@ private void handleQueryResponse(QueryStatusResponse<ResponseT> queryResponse)
       throw new IllegalStateException("Query status response missing X-Goog-Upload-Status header");
     }
     if (queryResponse.getUploadStatus() == ResumableUploadStatus.FINAL) {
-      progressTracker.onFinalized(buffer.getBufferBaseOffset() + buffer.getPayload().length);
+      long finalizedBytes =
+          Optional.ofNullable(queryResponse.getCommittedOffset())
+              .orElse(buffer.getBufferBaseOffset() + buffer.getPayload().length);
+      progressTracker.onFinalized(finalizedBytes);
       uploadResultFuture.set(queryResponse.getResponse());
       return;
     }

@@ -71,7 +71,7 @@
 
   private final Object lock = new Object();
 
-  private final ApiFuture<ResumableUploadSession> startFuture;
+  private final @Nullable ApiFuture<ResumableUploadSession> startFuture;
   private final UnaryCallable<ChunkUploadRequest, ChunkUploadResponse<ResponseT>>
       uploadChunkCallable;
   private final UnaryCallable<QueryStatusRequest, QueryStatusResponse<ResponseT>>
@@ -111,10 +111,12 @@
       ResumableUploadOptions options,
       ClientContext clientContext,
       ExponentialRetryAlgorithm recoveryAlgorithm) {
+    checkNotNull(startFuture, "startFuture must not be null");
     checkNotNull(payloadSupplier, "payloadSupplier must not be null");
     ResumableUploadFutureImpl<ResponseT> future =
         new ResumableUploadFutureImpl<>(
             startFuture,
+            /* sessionUrl= */ null,
             uploadChunkCallable,
             queryStatusCallable,
             options,
@@ -129,15 +131,51 @@
     return future;
   }
 
+  /**
+   * Creates and initiates a resumable upload future that resumes an existing upload session from
+   * {@code sessionUrl}.
+   *
+   * <p>The stream opened from {@code payloadSupplier} is managed by the returned future and will be
+   * closed automatically upon completion, failure, or cancellation.
+   */
+  static <ResponseT> ResumableUploadFutureImpl<ResponseT> createForResume(
+      String sessionUrl,
+      UnaryCallable<ChunkUploadRequest, ChunkUploadResponse<ResponseT>> uploadChunkCallable,
+      UnaryCallable<QueryStatusRequest, QueryStatusResponse<ResponseT>> queryStatusCallable,
+      InputStreamSupplier payloadSupplier,
+      ResumableUploadOptions options,
+      ClientContext clientContext,
+      ExponentialRetryAlgorithm recoveryAlgorithm) {
+    checkNotNull(sessionUrl, "sessionUrl must not be null");
+    checkNotNull(payloadSupplier, "payloadSupplier must not be null");
+    ResumableUploadFutureImpl<ResponseT> future =
+        new ResumableUploadFutureImpl<>(
+            /* startFuture= */ null,
+            sessionUrl,
+            uploadChunkCallable,
+            queryStatusCallable,
+            options,
+            clientContext,
+            recoveryAlgorithm,
+            ResumableUploadProgressTracker.createForResume(sessionUrl, options.getChunkSize()));
+    try {
+      future.resume(sessionUrl, payloadSupplier);
+    } catch (Throwable t) {
+      future.fail(t);
+    }
+    return future;
+  }
+
   private ResumableUploadFutureImpl(
-      ApiFuture<ResumableUploadSession> startFuture,
+      @Nullable ApiFuture<ResumableUploadSession> startFuture,
+      @Nullable String sessionUrl,
       UnaryCallable<ChunkUploadRequest, ChunkUploadResponse<ResponseT>> uploadChunkCallable,
       UnaryCallable<QueryStatusRequest, QueryStatusResponse<ResponseT>> queryStatusCallable,
       ResumableUploadOptions options,
       ClientContext clientContext,
       ExponentialRetryAlgorithm recoveryAlgorithm,
       ResumableUploadProgressTracker progressTracker) {
-    this.startFuture = checkNotNull(startFuture, "startFuture must not be null");
+    this.startFuture = startFuture;
     this.uploadChunkCallable =
         checkNotNull(uploadChunkCallable, "uploadChunkCallable must not be null");
     this.queryStatusCallable =
@@ -149,7 +187,12 @@
     this.executor = checkNotNull(clientContext.getExecutor(), "executor must not be null");
     this.recoveryAlgorithm = checkNotNull(recoveryAlgorithm, "recoveryAlgorithm must not be null");
     this.progressTracker = checkNotNull(progressTracker, "progressTracker must not be null");
-    this.inFlightFuture = startFuture;
+    if (sessionUrl != null) {
+      this.uploadSessionUrl = sessionUrl;
+      this.effectiveChunkSize = options.getChunkSize();
+    }
+    // Non-null placeholder marks the upload as active until the coordinator registers.
+    this.inFlightFuture = firstNonNull(startFuture, SettableApiFuture.create());
     ResumableUploadProgressListener progressListener = options.getProgressListener();
     if (progressListener != null) {
       progressTracker.addListener(progressListener, executor);
@@ -160,7 +203,7 @@
     setUpGlobalTimeout();
     this.payload = checkNotNull(payloadSupplier.get(), "payload supplier returned null");
     ApiFutures.addCallback(
-        startFuture,
+        checkNotNull(startFuture),
         new ApiFutureCallback<ResumableUploadSession>() {
           @Override
           public void onSuccess(ResumableUploadSession session) {
@@ -190,6 +233,16 @@
         MoreExecutors.directExecutor());
   }
 
+  private void resume(String sessionUrl, InputStreamSupplier payloadSupplier) throws IOException {
+    setUpGlobalTimeout();
+    this.payload = checkNotNull(payloadSupplier.get(), "payload supplier returned null");
+    ResumableUploadChunkCoordinator<ResponseT> coordinator =
+        createAndRegisterCoordinator(sessionUrl, options.getChunkSize());
+    if (coordinator != null) {
+      coordinator.resume();
+    }
+  }
+
   private void setUpGlobalTimeout() {
     Duration timeout = firstNonNull(options.getGlobalTimeout(), DEFAULT_GLOBAL_TIMEOUT);
     ScheduledFuture<?> timeoutFuture =

@@ -101,6 +101,16 @@ static ResumableUploadProgressTracker create() {
     return new ResumableUploadProgressTracker(ResumableUploadProgress.newBuilder().build());
   }
 
+  static ResumableUploadProgressTracker createForResume(String uploadUrl, int effectiveChunkSize) {
+    checkNotNull(uploadUrl, "uploadUrl must not be null");
+    return new ResumableUploadProgressTracker(
+        ResumableUploadProgress.newBuilder()
+            .setState(ResumableUploadProgress.STATE_RECOVERING)
+            .setUploadUrl(uploadUrl)
+            .setEffectiveChunkSize(effectiveChunkSize)
+            .build());
+  }
+
   void addListener(ResumableUploadProgressListener listener, Executor executor) {
     checkNotNull(listener, "listener must not be null");
     checkNotNull(executor, "executor must not be null");

@@ -33,6 +33,7 @@
 import static com.google.common.base.Preconditions.checkNotNull;
 
 import com.google.common.io.ByteStreams;
+import java.io.EOFException;
 import java.io.IOException;
 import java.io.InputStream;
 import java.util.Arrays;
@@ -52,6 +53,7 @@ final class RewindableStreamBuffer {
   private int payloadLength;
   private boolean isFinal;
   private boolean streamExhausted;
+  private boolean hasReadFromStream;
 
   RewindableStreamBuffer(InputStream inputStream, int chunkSize) {
     this.inputStream = checkNotNull(inputStream, "inputStream must not be null");
@@ -70,6 +72,7 @@ final class RewindableStreamBuffer {
    * @throws IOException if reading from the stream fails
    */
   void fill() throws IOException {
+    this.hasReadFromStream = true;
     this.bufferBaseOffset += payloadLength;
     this.payloadLength = ByteStreams.read(inputStream, buffer, 0, chunkSize);
     this.isFinal = (payloadLength < chunkSize);
@@ -81,15 +84,39 @@ void fill() throws IOException {
   /**
    * Realigns the buffer window to {@code committedOffset}.
    *
-   * <p>Compacts forward within the existing buffer to discard already-committed bytes, and then
-   * tops up the buffer to capacity from the underlying stream.
+   * <p>Before the first {@link #fill()}, skips the underlying stream forward from byte {@code 0} to
+   * {@code committedOffset} and fills the first chunk. Once data has been read, compacts forward
+   * within the existing buffer to discard already-committed bytes and tops up the buffer to
+   * capacity from the underlying stream.
    *
    * @param committedOffset the server's committed byte offset
-   * @throws IllegalStateException if {@code committedOffset} is below the buffer's base offset or
-   *     beyond the current buffer window
-   * @throws IOException if reading from the stream fails
+   * @throws IllegalStateException if {@code committedOffset} is negative, or (once data has been
+   *     read) below the buffer's base offset or beyond the current buffer window
+   * @throws IOException if skipping or reading from the stream fails
    */
   void realignTo(long committedOffset) throws IOException {
+    if (committedOffset < 0) {
+      throw new IllegalStateException(
+          String.format("Server committed offset %d must not be negative", committedOffset));
+    }
+    if (!hasReadFromStream) {
+      try {
+        ByteStreams.skipFully(inputStream, committedOffset);
+      } catch (EOFException e) {
+        EOFException ex =
+            new EOFException(
+                String.format(
+                    "Stream ended before reaching server committed offset %d; ensure the supplied"
+                        + " stream is positioned at byte 0",
+                    committedOffset));
+        ex.initCause(e);
+        throw ex;
+      }
+      this.bufferBaseOffset = committedOffset;
+      this.payloadLength = 0;
+      fill();
+      return;
+    }
     if (committedOffset < bufferBaseOffset) {
       throw new IllegalStateException(
           String.format(