feat(gax): add ResumableUploadFutureImpl state machine and ResumableUploadClient SPI

[blakeli0]

Description

This PR introduces the execution engine and state machine for GAX HTTP/JSON resumable uploads, stacked on top of PR #14052.

Changes

1. ResumableUploadFutureImpl: Non-blocking, stateful per-request state machine implementing ResumableUploadFuture<ResponseT>. Manages payload chunking, InputStream seeking, deadline enforcement, and error recovery.
2. ResumableUploadClient: Low-level transport Service Provider Interface (SPI) declaring startUploadCallable(), uploadChunkCallable(), and queryStatusCallable().
3. Value Objects: Added ResumableUploadSession, ChunkUploadRequest, and ChunkUploadResponse.
4. ResumableUploadCallable: Updated to instantiate and execute ResumableUploadFutureImpl.

[gemini-code-assist]

Code Review

This pull request introduces a new API-transport-independent resumable upload protocol implementation, including value objects (ChunkUploadRequest, ChunkUploadResponse), settings (ResumableUploadCallSettings), and a stateful future implementation (ResumableUploadFutureImpl) to manage chunking, stream offsets, and recovery. The review feedback highlights critical issues in ResumableUploadFutureImpl: a potential silent failure in the manual stream skipping logic, inefficient chunk reading due to partial reads from InputStream.read, and a race condition in cancel() that could allow uploads to continue in the background after cancellation. Using Guava's ByteStreams utility and synchronizing updates to the in-flight future are recommended to address these issues.

I am having trouble creating individual review comments. Click here to see my feedback.

sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadFutureImpl.java (135-146)

The current manual stream skipping loop can silently fail to skip the required number of bytes if payload.skip returns 0 or a negative value (which is common for certain stream types or when reaching EOF), leading to silent data corruption as the upload proceeds from an incorrect stream position.

Using Guava's ByteStreams.skipFully guarantees that exactly committedOffset bytes are skipped, throwing an EOFException if the stream ends prematurely, which is safely caught and propagates the failure to the future.

              if (committedOffset > 0) {
                ByteStreams.skipFully(payload, committedOffset);
              }

sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadFutureImpl.java (163-180)

Using InputStream.read(byte[]) does not guarantee that the entire buffer is filled even if more data is available (especially for network or buffered streams). This can result in uploading many small chunks, causing severe performance degradation.

Additionally, if we use Guava's ByteStreams.read to fully fill the buffer, any result where bytesRead < chunkSize indicates that EOF has been reached. This allows us to set isLast = true on the final non-empty chunk, saving an extra 0-byte HTTP round-trip at the end of the upload.

      int chunkSize = settings != null ? settings.getChunkSize() : 8 * 1024 * 1024;
      byte[] buffer = new byte[chunkSize];
      int bytesRead = ByteStreams.read(payload, buffer, 0, chunkSize);

      byte[] chunkData;
      if (bytesRead < chunkSize) {
        chunkData = new byte[bytesRead];
        System.arraycopy(buffer, 0, chunkData, 0, bytesRead);
      } else {
        chunkData = buffer;
      }

      ChunkUploadRequest chunkRequest =
          new ChunkUploadRequest(uploadSessionUrl, chunkData, committedOffset, -1L, bytesRead < chunkSize);

sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadFutureImpl.java (221-227)

There is a race condition between cancel() and the asynchronous callbacks starting the next steps (e.g., uploadNextChunk()). If cancel() is called, it cancels the current inFlightFuture. However, if a callback completes at the same time, it might initiate the next chunk upload and overwrite inFlightFuture with a new future that is never cancelled, causing the upload to continue silently in the background.

To prevent this, we should synchronize updates to inFlightFuture and check isCancelled() before starting any new asynchronous calls. Please also replace the direct assignments to this.inFlightFuture with setInFlightFuture(...) at lines 104, 126, and 184.

  private final Object lock = new Object();

  private void setInFlightFuture(ApiFuture<?> future) {
    synchronized (lock) {
      if (isCancelled()) {
        future.cancel(true);
        return;
      }
      this.inFlightFuture = future;
    }
  }

  @Override
  public boolean cancel(boolean mayInterruptIfRunning) {
    boolean cancelled = super.cancel(mayInterruptIfRunning);
    if (cancelled) {
      synchronized (lock) {
        if (inFlightFuture != null) {
          inFlightFuture.cancel(mayInterruptIfRunning);
        }
      }
    }
    return cancelled;
  }

sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadFutureImpl.java (37-38)

Import ByteStreams from Guava to simplify stream skipping and reading operations.

import com.google.common.base.Preconditions;
import com.google.common.io.ByteStreams;
import com.google.common.util.concurrent.MoreExecutors;