@@ -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);
 
@@ -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,15 @@ 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 advances the stream to
+   * that offset before resuming chunk transmission.
+   *
+   * <p>Because server chunk granularity is only returned in the initial {@code start} response,
+   * 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(