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;

@@ -49,51 +49,65 @@ protected ResumableUploadCallable() {}
   /**
    * Performs a new resumable upload asynchronously with options overrides and default call context.
    *
-   * <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.
+   * <p>Streams from the provided {@code payloadSupplier} are consumed asynchronously by the
+   * returned {@link ResumableUploadFuture} and will be closed automatically upon completion,
+   * failure, or cancellation.
    *
    * @param request the request message
-   * @param payload supplier of the data payload input stream to upload and close
+   * @param payloadSupplier supplier of the data payload input stream to upload and close
    * @param options request-level upload options overrides; may be {@code null}
    * @return future for tracking and controlling the upload
    */
   public ResumableUploadFuture<ResponseT> futureCall(
-      RequestT request, InputStreamSupplier payload, @Nullable ResumableUploadOptions options) {
-    return futureCall(request, payload, null, options);
+      RequestT request,
+      InputStreamSupplier payloadSupplier,
+      @Nullable ResumableUploadOptions options) {
+    return futureCall(request, payloadSupplier, null, options);
   }
 
   /**
    * Performs a new resumable upload asynchronously with call context and options overrides.
    *
-   * <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.
+   * <p>Streams from the provided {@code payloadSupplier} are consumed asynchronously by the
+   * returned {@link ResumableUploadFuture} and will be closed automatically upon completion,
+   * failure, or cancellation.
    *
    * @param request the request message
-   * @param payload supplier of the data payload input stream to upload and close
+   * @param payloadSupplier supplier of the data payload input stream to upload and close
    * @param context call context overrides; may be {@code null}
    * @param options request-level upload options overrides; may be {@code null}
    * @return future for tracking and controlling the upload
    */
   public abstract ResumableUploadFuture<ResponseT> futureCall(
       RequestT request,
-      InputStreamSupplier payload,
+      InputStreamSupplier payloadSupplier,
       @Nullable ApiCallContext context,
       @Nullable ResumableUploadOptions options);
 
   /**
    * Resumes an existing resumable upload session asynchronously using a saved session URL.
    *
-   * <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.
+   * <p>Resuming queries the server for the last committed byte offset and skips the stream forward
+   * to that offset before resuming chunk transmission. {@code payloadSupplier} 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 payloadSupplier} are consumed asynchronously by the
+   * returned {@link ResumableUploadFuture} and will be closed automatically upon completion,
+   * failure, or cancellation.
    *
    * @param sessionUrl the upload session URL
-   * @param payload supplier of the data payload input stream to upload and close
+   * @param payloadSupplier supplier of the data payload input stream to upload and close
    * @param options upload options overrides; may be {@code null}
    * @return future for tracking and controlling the upload
    */
   public abstract ResumableUploadFuture<ResponseT> resumeCall(
-      String sessionUrl, InputStreamSupplier payload, @Nullable ResumableUploadOptions options);
+      String sessionUrl,
+      InputStreamSupplier payloadSupplier,
+      @Nullable ResumableUploadOptions options);
 }
@@ -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;
@@ -62,7 +64,7 @@
 public class ResumableUploadCallableImpl<RequestT, ResponseT>
     extends ResumableUploadCallable<RequestT, ResponseT> {
 
-  private static final RetrySettings RETRY_SETTINGS =
+  static final RetrySettings RETRY_SETTINGS =
       RetrySettings.newBuilder()
           .setInitialRetryDelayDuration(Duration.ofSeconds(1))
           .setRetryDelayMultiplier(2.0)
@@ -81,7 +83,6 @@ public class ResumableUploadCallableImpl<RequestT, ResponseT>
       retryingUploadChunkCallable;
   private final UnaryCallable<QueryStatusRequest, QueryStatusResponse<ResponseT>>
       retryingQueryCallable;
-  private final ExponentialRetryAlgorithm recoveryAlgorithm;
 
   public ResumableUploadCallableImpl(
       ResumableUploadClient<RequestT, ResponseT> client,
@@ -96,18 +97,16 @@ public ResumableUploadCallableImpl(
     this.retryingQueryCallable =
         createRetryingCallable(
             client.queryStatusCallable(), ResumableUploadCommand.QUERY, clientContext);
-    this.recoveryAlgorithm =
-        new ExponentialRetryAlgorithm(RETRY_SETTINGS, clientContext.getClock());
   }
 
   @Override
   public ResumableUploadFuture<ResponseT> futureCall(
       RequestT request,
-      InputStreamSupplier payload,
+      InputStreamSupplier payloadSupplier,
       @Nullable ApiCallContext context,
       @Nullable ResumableUploadOptions options) {
     checkNotNull(request, "request must not be null");
-    checkNotNull(payload, "payload must not be null");
+    checkNotNull(payloadSupplier, "payloadSupplier must not be null");
     ResumableUploadOptions effectiveOptions = defaultOptions.merge(options);
     ApiCallContext effectiveCallContext = clientContext.getDefaultCallContext().merge(context);
 
@@ -122,16 +121,27 @@ public ResumableUploadFuture<ResponseT> futureCall(
         startFuture,
         retryingUploadChunkCallable,
         retryingQueryCallable,
-        payload,
+        payloadSupplier,
         effectiveOptions,
-        clientContext,
-        recoveryAlgorithm);
+        clientContext);
   }
 
   @Override
   public ResumableUploadFuture<ResponseT> resumeCall(
-      String sessionUrl, InputStreamSupplier payload, @Nullable ResumableUploadOptions options) {
-    throw new UnsupportedOperationException("Session resumption is not yet implemented.");
+      String sessionUrl,
+      InputStreamSupplier payloadSupplier,
+      @Nullable ResumableUploadOptions options) {
+    checkArgument(!Strings.isNullOrEmpty(sessionUrl), "sessionUrl must not be null or empty");
+    checkNotNull(payloadSupplier, "payloadSupplier must not be null");
+    ResumableUploadOptions effectiveOptions = defaultOptions.merge(options);
+
+    return ResumableUploadFutureImpl.createForResume(
+        sessionUrl,
+        retryingUploadChunkCallable,
+        retryingQueryCallable,
+        payloadSupplier,
+        effectiveOptions,
+        clientContext);
   }
 
   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;
     }