From 5563dbb95933aabef4780333afe7c72d96f2c305 Mon Sep 17 00:00:00 2001 From: whowes Date: Fri, 16 Oct 2026 19:51:29 +0000 Subject: [PATCH] feat(gax): implement resumable upload session resumption --- .../v1beta1/it/ITResumableUpload.java | 60 ++++++++ .../ResumableUploadQueryStatusCallable.java | 1 + .../api/gax/rpc/InputStreamSupplier.java | 5 +- .../api/gax/rpc/ResumableUploadCallable.java | 46 ++++-- .../gax/rpc/ResumableUploadCallableImpl.java | 32 ++-- .../rpc/ResumableUploadChunkCoordinator.java | 11 +- .../gax/rpc/ResumableUploadFutureImpl.java | 76 +++++++--- .../rpc/ResumableUploadProgressTracker.java | 10 ++ .../api/gax/rpc/RewindableStreamBuffer.java | 37 ++++- .../rpc/ResumableUploadCallableImplTest.java | 141 +++++++++++++++++- .../gax/rpc/RewindableStreamBufferTest.java | 34 +++++ 11 files changed, 394 insertions(+), 59 deletions(-) diff --git a/java-showcase/gapic-showcase/src/test/java/com/google/showcase/v1beta1/it/ITResumableUpload.java b/java-showcase/gapic-showcase/src/test/java/com/google/showcase/v1beta1/it/ITResumableUpload.java index 871374fbc565..d6862e279315 100644 --- a/java-showcase/gapic-showcase/src/test/java/com/google/showcase/v1beta1/it/ITResumableUpload.java +++ b/java-showcase/gapic-showcase/src/test/java/com/google/showcase/v1beta1/it/ITResumableUpload.java @@ -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 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 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 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); diff --git a/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/ResumableUploadQueryStatusCallable.java b/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/ResumableUploadQueryStatusCallable.java index 69d8042fd8b4..9d3c4eaecd0c 100644 --- a/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/ResumableUploadQueryStatusCallable.java +++ b/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/ResumableUploadQueryStatusCallable.java @@ -228,6 +228,7 @@ public void onClose(int statusCode, HttpJsonMetadata trailers) { future.set( QueryStatusResponse.newBuilder() .setUploadStatus(uploadStatus) + .setCommittedOffset(committedOffset) .setResponse(responseParser.parse(stream)) .build()); } else if (committedOffset != null) { diff --git a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/InputStreamSupplier.java b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/InputStreamSupplier.java index ceb6a9b8a347..3a0f8d9311a8 100644 --- a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/InputStreamSupplier.java +++ b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/InputStreamSupplier.java @@ -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. * *

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; diff --git a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadCallable.java b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadCallable.java index 1418d3fb3bbe..cce0b979d1e8 100644 --- a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadCallable.java +++ b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadCallable.java @@ -49,51 +49,65 @@ protected ResumableUploadCallable() {} /** * Performs a new resumable upload asynchronously with options overrides and default call context. * - *

Streams from the provided {@code payload} are consumed asynchronously by the returned {@link - * ResumableUploadFuture} and will be closed automatically upon completion, failure, or - * cancellation. + *

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 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. * - *

Streams from the provided {@code payload} are consumed asynchronously by the returned {@link - * ResumableUploadFuture} and will be closed automatically upon completion, failure, or - * cancellation. + *

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 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. * - *

Streams from the provided {@code payload} are consumed asynchronously by the returned {@link - * ResumableUploadFuture} and will be closed automatically upon completion, failure, or - * cancellation. + *

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. + * + *

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)}. + * + *

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 resumeCall( - String sessionUrl, InputStreamSupplier payload, @Nullable ResumableUploadOptions options); + String sessionUrl, + InputStreamSupplier payloadSupplier, + @Nullable ResumableUploadOptions options); } diff --git a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadCallableImpl.java b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadCallableImpl.java index 127678aa660f..b0bf9792f925 100644 --- a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadCallableImpl.java +++ b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadCallableImpl.java @@ -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 extends ResumableUploadCallable { - 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 retryingUploadChunkCallable; private final UnaryCallable> retryingQueryCallable; - private final ExponentialRetryAlgorithm recoveryAlgorithm; public ResumableUploadCallableImpl( ResumableUploadClient 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 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 futureCall( startFuture, retryingUploadChunkCallable, retryingQueryCallable, - payload, + payloadSupplier, effectiveOptions, - clientContext, - recoveryAlgorithm); + clientContext); } @Override public ResumableUploadFuture 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 UnaryCallable createRetryingCallable( diff --git a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadChunkCoordinator.java b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadChunkCoordinator.java index c84a42d81a32..9d74447eefff 100644 --- a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadChunkCoordinator.java +++ b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadChunkCoordinator.java @@ -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 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; } diff --git a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadFutureImpl.java b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadFutureImpl.java index defc964090c3..6ebb73c39b21 100644 --- a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadFutureImpl.java +++ b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadFutureImpl.java @@ -71,7 +71,6 @@ final class ResumableUploadFutureImpl implements ResumableUploadFutur private final Object lock = new Object(); - private final ApiFuture startFuture; private final UnaryCallable> uploadChunkCallable; private final UnaryCallable> @@ -109,20 +108,51 @@ static ResumableUploadFutureImpl create( UnaryCallable> queryStatusCallable, InputStreamSupplier payloadSupplier, ResumableUploadOptions options, - ClientContext clientContext, - ExponentialRetryAlgorithm recoveryAlgorithm) { + ClientContext clientContext) { + checkNotNull(startFuture, "startFuture must not be null"); checkNotNull(payloadSupplier, "payloadSupplier must not be null"); ResumableUploadFutureImpl future = new ResumableUploadFutureImpl<>( - startFuture, + /* sessionUrl= */ null, uploadChunkCallable, queryStatusCallable, options, - clientContext, - recoveryAlgorithm, - ResumableUploadProgressTracker.create()); + clientContext); try { - future.start(payloadSupplier); + future.start(startFuture, payloadSupplier); + } catch (Throwable t) { + future.fail(t); + } + return future; + } + + /** + * Creates and initiates a resumable upload future that resumes an existing upload session from + * {@code sessionUrl}. + * + *

The stream opened from {@code payloadSupplier} is managed by the returned future and will be + * closed automatically upon completion, failure, or cancellation. + */ + static ResumableUploadFutureImpl createForResume( + String sessionUrl, + UnaryCallable> uploadChunkCallable, + UnaryCallable> queryStatusCallable, + InputStreamSupplier payloadSupplier, + ResumableUploadOptions options, + ClientContext clientContext) { + checkNotNull(sessionUrl, "sessionUrl must not be null"); + checkNotNull(payloadSupplier, "payloadSupplier must not be null"); + ResumableUploadFutureImpl future = + new ResumableUploadFutureImpl<>( + sessionUrl, uploadChunkCallable, queryStatusCallable, options, clientContext); + try { + future.setUpGlobalTimeout(); + future.payload = checkNotNull(payloadSupplier.get(), "payload supplier returned null"); + ResumableUploadChunkCoordinator coordinator = + future.createAndRegisterCoordinator(sessionUrl, options.getChunkSize()); + if (coordinator != null) { + coordinator.resume(); + } } catch (Throwable t) { future.fail(t); } @@ -130,14 +160,11 @@ static ResumableUploadFutureImpl create( } private ResumableUploadFutureImpl( - ApiFuture startFuture, + @Nullable String sessionUrl, UnaryCallable> uploadChunkCallable, UnaryCallable> queryStatusCallable, ResumableUploadOptions options, - ClientContext clientContext, - ExponentialRetryAlgorithm recoveryAlgorithm, - ResumableUploadProgressTracker progressTracker) { - this.startFuture = checkNotNull(startFuture, "startFuture must not be null"); + ClientContext clientContext) { this.uploadChunkCallable = checkNotNull(uploadChunkCallable, "uploadChunkCallable must not be null"); this.queryStatusCallable = @@ -147,16 +174,31 @@ private ResumableUploadFutureImpl( checkNotNull(clientContext, "clientContext must not be null"); this.callContext = clientContext.getDefaultCallContext(); 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; + this.recoveryAlgorithm = + new ExponentialRetryAlgorithm( + ResumableUploadCallableImpl.RETRY_SETTINGS, clientContext.getClock()); + if (sessionUrl != null) { + this.uploadSessionUrl = sessionUrl; + this.effectiveChunkSize = options.getChunkSize(); + this.progressTracker = + ResumableUploadProgressTracker.createForResume(sessionUrl, options.getChunkSize()); + } else { + this.progressTracker = ResumableUploadProgressTracker.create(); + } + // Non-null placeholder marks the upload as active until the coordinator registers. + this.inFlightFuture = SettableApiFuture.create(); ResumableUploadProgressListener progressListener = options.getProgressListener(); if (progressListener != null) { progressTracker.addListener(progressListener, executor); } } - private void start(InputStreamSupplier payloadSupplier) throws IOException { + private void start( + ApiFuture startFuture, InputStreamSupplier payloadSupplier) + throws IOException { + synchronized (lock) { + this.inFlightFuture = startFuture; + } setUpGlobalTimeout(); this.payload = checkNotNull(payloadSupplier.get(), "payload supplier returned null"); ApiFutures.addCallback( diff --git a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadProgressTracker.java b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadProgressTracker.java index df2887effe20..b5d9121f2d48 100644 --- a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadProgressTracker.java +++ b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadProgressTracker.java @@ -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"); diff --git a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/RewindableStreamBuffer.java b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/RewindableStreamBuffer.java index 0030f0282793..06db44ca944b 100644 --- a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/RewindableStreamBuffer.java +++ b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/RewindableStreamBuffer.java @@ -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}. * - *

Compacts forward within the existing buffer to discard already-committed bytes, and then - * tops up the buffer to capacity from the underlying stream. + *

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( diff --git a/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/ResumableUploadCallableImplTest.java b/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/ResumableUploadCallableImplTest.java index 6bcfc9d234a2..f1b98ecfb1f6 100644 --- a/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/ResumableUploadCallableImplTest.java +++ b/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/ResumableUploadCallableImplTest.java @@ -465,13 +465,6 @@ void testUploadCallable_withContextAndSettings_appliesBoth() throws Exception { assertThat(future.get()).isEqualTo("done-both"); } - @Test - void testResumeCall_throwsUnsupportedOperationException() { - assertThrows( - UnsupportedOperationException.class, - () -> callable.resumeCall("https://upload.url/session", payloadOf("data"), null)); - } - @Test void testChunkRetry_transientFailureThenSuccess_retriesAndSucceeds() throws Exception { stubStartSession("https://upload.url/chunk-retry-ok"); @@ -1383,6 +1376,140 @@ public void close() throws IOException { .contains("stream close error"); } + @Test + void testResumeCall_partialOffset_resumesAndCompletes() throws Exception { + String sessionUrl = "https://upload.url/resume-partial"; + when(mockQueryCallable.futureCall(eq(QueryStatusRequest.create(sessionUrl)), any())) + .thenReturn( + ApiFutures.immediateFuture( + createQueryResponse(8L, null, ResumableUploadStatus.ACTIVE))); + when(mockChunkCallable.futureCall(any(ChunkUploadRequest.class), any())) + .thenReturn( + ApiFutures.immediateFuture( + ChunkUploadResponse.create(ResumableUploadStatus.ACTIVE, null))) + .thenReturn( + ApiFutures.immediateFuture( + ChunkUploadResponse.create(ResumableUploadStatus.FINAL, "resumed-done"))); + + RecordingProgressListener listener = new RecordingProgressListener(); + ResumableUploadOptions options = + defaultSettings.toBuilder().setProgressListener(listener).build(); + + TrackableStream stream = new TrackableStream("0123456789abcdefghij"); + ResumableUploadFuture future = callable.resumeCall(sessionUrl, () -> stream, options); + + assertThat(future.getUploadSessionUrl()).isEqualTo(sessionUrl); + assertThat(future.getEffectiveChunkSize()).isEqualTo(8); + assertThat(future.get()).isEqualTo("resumed-done"); + assertThat(stream.closed).isTrue(); + verifyNoInteractions(mockStartCallable); + + ArgumentCaptor captor = ArgumentCaptor.forClass(ChunkUploadRequest.class); + verify(mockChunkCallable, times(2)).futureCall(captor.capture(), any()); + List chunks = captor.getAllValues(); + assertChunk(chunks.get(0), 8, 8, false); + assertChunk(chunks.get(1), 16, 4, true); + assertThat(new String(chunks.get(1).getPayload(), StandardCharsets.UTF_8)).isEqualTo("ghij"); + + assertThat(listener.awaitTerminal()) + .containsExactly( + createProgress(ResumableUploadProgress.STATE_RECOVERING, 0L, sessionUrl, 8), + createProgress(ResumableUploadProgress.STATE_OFFSET_RECEIVED, 8L, sessionUrl, 8), + createProgress(ResumableUploadProgress.STATE_UPLOADING, 16L, sessionUrl, 8), + createProgress(ResumableUploadProgress.STATE_FINALIZED, 20L, sessionUrl, 8)) + .inOrder(); + } + + @Test + void testResumeCall_alreadyFinalizedOnServer_completesImmediately() throws Exception { + String sessionUrl = "https://upload.url/resume-already-final"; + when(mockQueryCallable.futureCall(eq(QueryStatusRequest.create(sessionUrl)), any())) + .thenReturn( + ApiFutures.immediateFuture( + createQueryResponse(null, "already-finalized", ResumableUploadStatus.FINAL))); + + ResumableUploadFuture future = + callable.resumeCall(sessionUrl, payloadOf("0123456789abcdefghij"), null); + + assertThat(future.get()).isEqualTo("already-finalized"); + verifyNoInteractions(mockStartCallable); + verifyNoInteractions(mockChunkCallable); + } + + @Test + void testResumeCall_initialQueryTransientFailure_retriesAndResumes() throws Exception { + String sessionUrl = "https://upload.url/resume-query-retry"; + when(mockQueryCallable.futureCall(eq(QueryStatusRequest.create(sessionUrl)), any())) + .thenReturn( + ApiFutures.immediateFailedFuture(createApiException(503, StatusCode.Code.UNAVAILABLE))) + .thenReturn( + ApiFutures.immediateFuture( + createQueryResponse(8L, null, ResumableUploadStatus.ACTIVE))); + when(mockChunkCallable.futureCall(any(ChunkUploadRequest.class), any())) + .thenReturn( + ApiFutures.immediateFuture( + ChunkUploadResponse.create(ResumableUploadStatus.FINAL, "resumed-after-503"))); + + ResumableUploadFuture future = + callable.resumeCall(sessionUrl, payloadOf("0123456789abcd"), null); + + assertThat(future.get()).isEqualTo("resumed-after-503"); + verify(mockQueryCallable, times(2)).futureCall(any(), any()); + ArgumentCaptor captor = ArgumentCaptor.forClass(ChunkUploadRequest.class); + verify(mockChunkCallable, times(1)).futureCall(captor.capture(), any()); + assertChunk(captor.getValue(), 8, 6, true); + assertThat(new String(captor.getValue().getPayload(), StandardCharsets.UTF_8)) + .isEqualTo("89abcd"); + } + + @Test + void testResumeCall_withSubsequentRecovery_realignsWithinChunkWindow() throws Exception { + String sessionUrl = "https://upload.url/resume-then-recover"; + // Initial resume query reports 8 bytes committed; mid-upload recovery reports 12 bytes + when(mockQueryCallable.futureCall(eq(QueryStatusRequest.create(sessionUrl)), any())) + .thenReturn( + ApiFutures.immediateFuture(createQueryResponse(8L, null, ResumableUploadStatus.ACTIVE))) + .thenReturn( + ApiFutures.immediateFuture( + createQueryResponse(12L, null, ResumableUploadStatus.ACTIVE))); + when(mockChunkCallable.futureCall(any(ChunkUploadRequest.class), any())) + .thenReturn( + ApiFutures.immediateFailedFuture( + createApiException(400, StatusCode.Code.INVALID_ARGUMENT))) + .thenReturn( + ApiFutures.immediateFuture( + ChunkUploadResponse.create(ResumableUploadStatus.FINAL, "recovered-after-resume"))); + + ResumableUploadFuture future = + callable.resumeCall(sessionUrl, payloadOf("0123456789abcdefgh"), null); + + assertThat(future.get()).isEqualTo("recovered-after-resume"); + ArgumentCaptor captor = ArgumentCaptor.forClass(ChunkUploadRequest.class); + verify(mockChunkCallable, times(2)).futureCall(captor.capture(), any()); + List chunks = captor.getAllValues(); + assertChunk(chunks.get(0), 8, 8, false); + assertChunk(chunks.get(1), 12, 6, true); + assertThat(new String(chunks.get(1).getPayload(), StandardCharsets.UTF_8)).isEqualTo("cdefgh"); + } + + @Test + void testResumeCall_failure_includesSessionUrlAndEffectiveChunkSize() { + String sessionUrl = "https://upload.url/resume-short-stream"; + when(mockQueryCallable.futureCall(eq(QueryStatusRequest.create(sessionUrl)), any())) + .thenReturn( + ApiFutures.immediateFuture( + createQueryResponse(16L, null, ResumableUploadStatus.ACTIVE))); + + TrackableStream shortStream = new TrackableStream("01234567"); + ResumableUploadFuture future = callable.resumeCall(sessionUrl, () -> shortStream, null); + + ExecutionException ex = assertThrows(ExecutionException.class, future::get); + assertThat(ex.getCause()).isInstanceOf(UnknownException.class); + assertThat(ex.getCause().getMessage()).contains(sessionUrl); + assertThat(ex.getCause().getMessage()).contains("effective chunk size: 8"); + assertThat(shortStream.closed).isTrue(); + } + private static class HttpStatusStatusCode implements StatusCode { private final int httpStatus; private final StatusCode.Code code; diff --git a/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/RewindableStreamBufferTest.java b/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/RewindableStreamBufferTest.java index 6ca4b8020b54..1aceab386cc8 100644 --- a/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/RewindableStreamBufferTest.java +++ b/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/RewindableStreamBufferTest.java @@ -33,6 +33,7 @@ import static org.junit.jupiter.api.Assertions.assertThrows; import java.io.ByteArrayInputStream; +import java.io.EOFException; import java.io.FilterInputStream; import java.io.IOException; import java.io.InputStream; @@ -205,4 +206,37 @@ public int read(byte[] b, int off, int len) throws IOException { assertThat(buffer.isFinal()).isFalse(); assertThat(new String(buffer.getPayload(), StandardCharsets.UTF_8)).isEqualTo("01234567"); } + + @Test + void testRealignTo_beforeFirstFill_skipsToCommittedOffset() throws IOException { + byte[] data = "0123456789ABCDEFGHIJ".getBytes(StandardCharsets.UTF_8); + RewindableStreamBuffer buffer = new RewindableStreamBuffer(new ByteArrayInputStream(data), 8); + + buffer.realignTo(10L); + assertThat(buffer.getBufferBaseOffset()).isEqualTo(10L); + assertThat(buffer.getPayload()).hasLength(8); + assertThat(buffer.isFinal()).isFalse(); + assertThat(new String(buffer.getPayload(), StandardCharsets.UTF_8)).isEqualTo("ABCDEFGH"); + } + + @Test + void testRealignTo_beforeFirstFill_streamShorterThanOffset_throwsEofException() { + byte[] data = "01234".getBytes(StandardCharsets.UTF_8); + RewindableStreamBuffer buffer = new RewindableStreamBuffer(new ByteArrayInputStream(data), 8); + + EOFException exception = assertThrows(EOFException.class, () -> buffer.realignTo(10L)); + assertThat(exception).hasMessageThat().contains("10"); + } + + @Test + void testRealignTo_beforeFirstFill_zeroOffset_fillsFromStart() throws IOException { + byte[] data = "0123456789".getBytes(StandardCharsets.UTF_8); + RewindableStreamBuffer buffer = new RewindableStreamBuffer(new ByteArrayInputStream(data), 8); + + buffer.realignTo(0L); + assertThat(buffer.getBufferBaseOffset()).isEqualTo(0L); + assertThat(buffer.getPayload()).hasLength(8); + assertThat(buffer.isFinal()).isFalse(); + assertThat(new String(buffer.getPayload(), StandardCharsets.UTF_8)).isEqualTo("01234567"); + } }