POC Option 1B: Specialized REST Upload Stub

[blakeli0]

This Proof of Concept implements an architectural enhancement over Option 1 by introducing a specialized upload-only stub: HttpJsonEchoResumableUploadStub.

Key Features

• Zero Background Overhead in gRPC: Instead of initializing a full HttpJsonEchoStub (which instantiates REST callables, method descriptors, and serializers for all standard RPCs in the service), this specialized stub implements ONLY the Scotty resumable upload RPCs.
• Unified & Symmetric Delegation: Both GrpcEchoStub and HttpJsonEchoStub instantiate this specialized upload stub in their constructor and delegate resumableUploadCallable() directly to it.
• Shared vs Secondary Transport Handling:
  • In HttpJsonEchoStub, it initializes the upload stub using its existing primary ClientContext without creating a secondary transport connection.
  • In GrpcEchoStub, it uses ClientContext.withTransportChannelProvider to clone credentials, headers, and executors onto a secondary HTTP/JSON transport channel.
• Clean Separation of Concerns: Keeps all REST method descriptors, routing paths, and JSON serializers out of the gRPC stub classes.

[gemini-code-assist]

Code Review

This pull request introduces support for the Resumable Upload Protocol (RUP) in gax-java and GAPIC-generated code, adding new core abstractions (such as ResumableUploadCallable and ResumableUploadRequest) and implementing the HTTP/JSON transport layer state machine. The review feedback highlights several critical areas for improvement in the RUP implementation, particularly in HttpJsonResumableUploadCall.java. Key recommendations include resolving an InputStream resource leak, avoiding redundant JSON parsing/serialization, using response.close() instead of response.disconnect() to enable connection reuse, capping timeout values to prevent integer overflow, properly handling thread interruption, and caching the HttpRequestFactory. Additionally, the reviewer suggests initializing the upload stub unconditionally in GrpcEchoStub to support credential-free local testing and avoiding a redundant double-close of the transport channel in HttpJsonEchoStub.

[gemini-code-assist]

The InputStream opened via uploadRequest.getStreamProvider().get() is never closed when the state machine completes successfully or fails with an exception, leading to a resource leak. Additionally, if skipFully fails to skip enough bytes (e.g., because the stream is shorter than the offset), it throws an IOException which is categorized as CATEGORY_1_TRANSIENT, causing the state machine to retry indefinitely with backoff. We should wrap the stream usage in a try-finally block to ensure it is always closed, and throw an IllegalStateException (treated as CATEGORY_3_FATAL) if the stream is too short.

    InputStream stream = null;
    try {
      stream = uploadRequest.getStreamProvider().get();
      long streamPosition = 0;

      List<BufferedChunk> cache = new ArrayList<>();

      // Phase 2 & 3 Loop: Transmit Chunks & Query Recovery
      while (true) {
        try {
          checkDeadline(deadline);

          // Find chunk in cache or read from stream
          BufferedChunk chunk = null;
          for (BufferedChunk cached : cache) {
            if (cached.offset == offset) {
              chunk = cached;
              break;
            }
          }

          if (chunk == null) {
            // Read from stream
            if (streamPosition != offset) {
              if (stream != null) {
                stream.close();
              }
              stream = uploadRequest.getStreamProvider().get();
              long skipped = skipFully(stream, offset);
              if (skipped < offset) {
                throw new IllegalStateException("Failed to skip stream bytes to offset: " + offset);
              }
              streamPosition = offset;
            }

            byte[] buffer = new byte[adjustedChunkSize];
            int bytesRead = readFully(stream, buffer, adjustedChunkSize);
            if (bytesRead > 0) {
              chunk = new BufferedChunk(offset, buffer, bytesRead);
              cache.add(chunk);
              if (cache.size() > 2) {
                cache.remove(0);
              }
              streamPosition += bytesRead;
            }
          }

          if (chunk == null) {
            // Stream was empty or exact chunk multiple and fully uploaded.
            // Send finalize only
            return sendFinalizeOnly(uploadUrl, offset, deadline);
          }

          // Check if this is the last chunk
          boolean isEof = (chunk.length < adjustedChunkSize);

          if (isEof) {
            // Send upload, finalize for the last chunk
            return sendUploadFinalize(uploadUrl, chunk.offset, chunk.data, chunk.length, deadline);
          }

          // Send intermediate chunk (upload command)
          sendChunk(uploadUrl, chunk.offset, chunk.data, chunk.length, deadline);

          // Successful chunk transmission! Update offset to next chunk
          offset = chunk.offset + chunk.length;
          attempt = 0; // Reset backoff attempts on progress

        } catch (UploadAlreadyFinalizedException uafe) {
          updateProgress(
              uploadRequest.getTotalBytes() > 0 ? uploadRequest.getTotalBytes() : offset,
              ResumableUploadProgressListener.State.COMPLETED);
          return (ResponseT) uafe.getResponse();
        } catch (Exception e) {
          checkDeadline(deadline);
          ErrorCategory category = getErrorCategory(e);

          if (category == ErrorCategory.CATEGORY_2_MISMATCH) {
            logger.log(Level.WARNING, "State mismatch detected. Triggering recovery...", e);
            updateProgress(offset, ResumableUploadProgressListener.State.RECOVERING);

            long serverOffset = 0;
            try {
              serverOffset = recoverOffset(uploadUrl, deadline);
            } catch (UploadAlreadyFinalizedException uafe) {
              updateProgress(
                  uploadRequest.getTotalBytes() > 0 ? uploadRequest.getTotalBytes() : offset,
                  ResumableUploadProgressListener.State.COMPLETED);
              return (ResponseT) uafe.getResponse();
            }

            logger.log(Level.INFO, "Recovery completed. Server received bytes: {0}", serverOffset);

            if (serverOffset == previousOffset) {
              attempt++;
              long delayMs = calculateBackoff(attempt);
              sleep(delayMs);
            } else {
              attempt = 0;
              previousOffset = serverOffset;
            }

            offset = serverOffset;
            // Loop will handle finding the chunk in cache or seeking/recreating the stream!
          } else if (category == ErrorCategory.CATEGORY_1_TRANSIENT) {
            attempt++;
            long delayMs = calculateBackoff(attempt);
            logger.log(
                Level.WARNING,
                "Transient error. Backing off for {0} ms (attempt {1})",
                new Object[] {delayMs, attempt});
            sleep(delayMs);
          } else {
            updateProgress(offset, ResumableUploadProgressListener.State.FAILED);
            throw e; // Fatal, bubble up
          }
        }
      }
    } finally {
      if (stream != null) {
        try {
          stream.close();
        } catch (IOException e) {
          logger.log(Level.WARNING, "Failed to close stream", e);
        }
      }
    }

[gemini-code-assist]

In startSession, requestBody (which is already a serialized JSON string) is parsed into a GenericData object and then re-serialized using JsonHttpContent. This is highly inefficient and consumes unnecessary memory and CPU. We should use ByteArrayContent.fromString to send the raw JSON string directly.

    if (!Strings.isNullOrEmpty(requestBody)) {
      initialContent = ByteArrayContent.fromString("application/json; charset=utf-8", requestBody);
    } else {
      initialContent = new EmptyContent();
    }

[gemini-code-assist]

In the finally blocks of all HTTP calls (such as startSession, sendChunk, sendUploadFinalize, sendFinalizeOnly, and recoverOffset), response.disconnect() is called. This closes the underlying socket and prevents connection reuse (keep-alive) for subsequent chunk uploads, causing a new TCP and TLS handshake for every single chunk. We should use response.close() instead to release the connection back to the pool.

    } finally {
      if (response != null) {
        try {
          response.close();
        } catch (IOException e) {
          logger.log(Level.WARNING, "Failed to close response", e);
        }
      }
    }

[gemini-code-assist]

In configureTimeouts, remainingMs is cast directly to (int) remainingMs. If the deadline is very far in the future (e.g., configured by the user to be very large), this can overflow/underflow and result in negative or extremely small timeouts. We should cap the timeout at Integer.MAX_VALUE.

Suggested change

	private void configureTimeouts(final HttpRequest request, final Instant deadline) {
	long remainingMs = Duration.between(Instant.now(), deadline).toMillis();
	if (remainingMs <= 0) {
	remainingMs = 1; // force timeout
	}
	request.setConnectTimeout((int) remainingMs);
	request.setReadTimeout((int) remainingMs);
	}
	private void configureTimeouts(final HttpRequest request, final Instant deadline) {
	long remainingMs = Duration.between(Instant.now(), deadline).toMillis();
	if (remainingMs <= 0) {
	remainingMs = 1; // force timeout
	}
	int timeoutMs = (remainingMs > Integer.MAX_VALUE) ? Integer.MAX_VALUE : (int) remainingMs;
	request.setConnectTimeout(timeoutMs);
	request.setReadTimeout(timeoutMs);
	}

[gemini-code-assist]

In sleep, if InterruptedException is caught, the interrupted status is restored, but the loop continues. This ignores the interruption request and can lead to unresponsive cancellation. We should abort the upload by throwing a RuntimeException or propagating the interruption.

Suggested change

	private void sleep(final long ms) {
	try {
	Thread.sleep(ms);
	} catch (InterruptedException e) {
	Thread.currentThread().interrupt();
	}
	}
	private void sleep(final long ms) {
	try {
	Thread.sleep(ms);
	} catch (InterruptedException e) {
	Thread.currentThread().interrupt();
	throw new RuntimeException("Upload interrupted during backoff sleep", e);
	}
	}

References

1. In Java, do not swallow InterruptedException. When catching it, restore the thread's interrupted status by calling Thread.currentThread().interrupt() and handle the interruption appropriately, such as by throwing a relevant exception (e.g., SpannerException) to signal that the operation cannot proceed.

[gemini-code-assist]

normalizeEndpoint ensures the endpoint ends with /, and path starts with /resumable/upload. Concatenating them directly (normalizeEndpoint(endpoint) + path) results in a double slash (e.g., https://localhost//resumable/upload/...), which can cause issues with strict HTTP proxies or load balancers. We should normalize the path to avoid double slashes.

    String path = "/resumable/upload" + requestFormatter.getPath(uploadRequest.getRequest());
    String normalizedEndpoint = normalizeEndpoint(endpoint);
    if (normalizedEndpoint.endsWith("/") && path.startsWith("/")) {
      path = path.substring(1);
    }
    GenericUrl url = new GenericUrl(normalizedEndpoint + path);

[gemini-code-assist]

GrpcEchoStub only initializes uploadStub if clientContext.getCredentials() != null, whereas HttpJsonEchoStub initializes it unconditionally. This prevents using resumable uploads in local testing or emulator environments where credentials are null. We should initialize uploadStub unconditionally.

    TransportChannelProvider httpJsonProvider =
        EchoStubSettings.defaultHttpJsonTransportProviderBuilder().build();
    ClientContext backgroundHttpContext =
        clientContext.withTransportChannelProvider(httpJsonProvider);
    this.uploadStub = HttpJsonEchoResumableUploadStub.create(settings, backgroundHttpContext);

[gemini-code-assist]

HttpJsonEchoStub shares its primary transport channel with uploadStub. When HttpJsonEchoStub.close() is called, it closes the primary channel via backgroundResources.close(), and then calls uploadStub.close(), which closes the same primary channel again. This double-close of the primary transport channel is redundant and can be avoided by not calling uploadStub.close() in HttpJsonEchoStub.close().

  public final void close() {
    try {
      backgroundResources.close();
    } catch (RuntimeException e) {
      throw e;
    } catch (Exception e) {
      throw new IllegalStateException("Failed to close resource", e);
    }

[gemini-code-assist]

Creating a new HttpRequestFactory on every single chunk upload is inefficient. Since callOptions and httpTransport are final and passed to the constructor, the HttpRequestFactory will always be the same for a given HttpJsonResumableUploadCall instance. We should initialize and cache it in a private final field in the constructor.