From 386ec3709c07db91f6bd00aabd119ca5f25af152 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 +++++ .../src/test/resources/compliance_suite.json | 240 ------------------ .../api/gax/rpc/InputStreamSupplier.java | 5 +- .../api/gax/rpc/ResumableUploadCallable.java | 9 + .../gax/rpc/ResumableUploadCallableImpl.java | 15 +- .../rpc/ResumableUploadChunkCoordinator.java | 37 ++- .../gax/rpc/ResumableUploadFutureImpl.java | 165 ++++++++---- .../rpc/ResumableUploadProgressTracker.java | 18 +- .../api/gax/rpc/RewindableStreamBuffer.java | 32 ++- .../rpc/ResumableUploadCallableImplTest.java | 115 ++++++++- .../ResumableUploadProgressTrackerTest.java | 12 +- .../gax/rpc/RewindableStreamBufferTest.java | 46 +++- 12 files changed, 421 insertions(+), 333 deletions(-) delete mode 100644 java-showcase/gapic-showcase/src/test/resources/compliance_suite.json 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/java-showcase/gapic-showcase/src/test/resources/compliance_suite.json b/java-showcase/gapic-showcase/src/test/resources/compliance_suite.json deleted file mode 100644 index a458819b3504..000000000000 --- a/java-showcase/gapic-showcase/src/test/resources/compliance_suite.json +++ /dev/null @@ -1,240 +0,0 @@ -{ - "group": [ - { - "name": "Fully working conversions, no resources", - "rpcs": ["Compliance.RepeatDataBody", "Compliance.RepeatDataBodyPut", "Compliance.RepeatDataBodyPatch", "Compliance.RepeatDataQuery", "Compliance.RepeatDataSimplePath", "Compliance.RepeatDataBodyInfo"], - "requests": [ - { - "name": "Basic data types", - "serverVerify": true, - "info": { - "fString": "Hello", - "fInt32": -1, - "fSint32" : -2, - "fSfixed32": -3, - "fUint32": 5, - "fFixed32": 7, - "fInt64": -11, - "fSint64": -13, - "fSfixed64": -17, - "fUint64": 19, - "fFixed64":23, - "fDouble": -29e4, - "fFloat": -31, - "fBool": true, - "fKingdom": "ANIMALIA", - - "pString": "Goodbye", - "pInt32": -37, - "pDouble": -41.43, - "pBool": true, - "pKingdom": "PLANTAE", - - "fChild": { - "fString": "second/bool/salutation" - } - }, - "fInt32": -10, - "fInt64": -110, - "fDouble": -54e4, - - "pInt32": -47, - "pInt64": -477, - "pDouble": -61.73 - }, - { - "name": "Basic types, no optional fields", - "serverVerify": true, - "info": { - "fString": "Hello", - "fInt32": -1, - "fSint32" : -2, - "fSfixed32": -3, - "fUint32": 5, - "fFixed32": 7, - "fInt64": -11, - "fSint64": -13, - "fSfixed64": -17, - "fUint64": 19, - "fFixed64":23, - "fDouble": -29e4, - "fFloat": -31, - "fBool": true, - "fKingdom": "ANIMALIA", - - "fChild": { - "fString": "second/bool/salutation" - } - } - }, - - { - "name": "Zero values for non-string fields", - "serverVerify": true, - "info": { - "fString": "Hello", - "fInt32": 0, - "fSint32" : 0, - "fSfixed32": 0, - "fUint32": 0, - "fFixed32": 0, - "fInt64": 0, - "fSint64": 0, - "fSfixed64": 0, - "fUint64": 0, - "fFixed64": 0, - "fDouble": 0, - "fFloat": 0, - "fBool": false, - "fKingdom": "LIFE_KINGDOM_UNSPECIFIED", - - "pString": "Goodbye", - "pInt32": 0, - "pDouble": 0, - "pBool": false, - "pKingdom": "LIFE_KINGDOM_UNSPECIFIED" - } - }, - { - "name": "Extreme values", - "serverVerify": true, - "info": { - "fString": "non-ASCII+non-printable string ☺ → ← \"\\\/\b\f\r\t\u1234 works, not newlines yet", - "fInt32": 2147483647, - "fSint32" : 2147483647, - "fSfixed32": 2147483647, - "fUint32": 4294967295, - "fFixed32": 4294967295, - "fInt64": "9223372036854775807", - "fSint64": "9223372036854775807", - "fSfixed64": "9223372036854775807", - "fUint64": "18446744073709551615", - "fFixed64": "18446744073709551615", - "fDouble": 1.797693134862315708145274237317043567981e+308, - "fFloat": 3.40282346638528859811704183484516925440e+38, - "fBool": false, - - "pString": "Goodbye", - "pInt32": 2147483647, - "pDouble": 1.797693134862315708145274237317043567981e+308, - "pBool": false - } - }, - { - "name": "Strings with spaces", - "serverVerify": true, - "info": { - "fString": "Hello there" - } - }, - { - "name": "Strings with quotes", - "serverVerify": true, - "info": { - "fString": "Hello \"You\"" - } - }, - { - "name": "Strings with percents", - "serverVerify": true, - "info": { - "fString": "Hello 100%" - } - } - ] - }, - { - "name": "Fully working conversions, resources", - "rpcs": ["Compliance.RepeatDataBody", "Compliance.RepeatDataBodyPut", "Compliance.RepeatDataBodyPatch", "Compliance.RepeatDataQuery"], - "requests": [ - { - "name": "Strings with slashes and values that resemble subsequent resource templates", - "serverVerify": true, - "info": { - "fString": "first/hello/second/greetings", - "pBool": true, - - "fChild": { - "fString": "second/zzz/bool/true" - } - } - } - ] - }, - { - "name": "Binding selection testing", - "rpcs": ["Compliance.RepeatDataPathResource"], - "requests": [ - { - "name": "Binding testing baseline no Uri verification", - "serverVerify": true, - "info": { - "fString": "first/hello", - "pBool": true, - - "fChild": { - "fString": "second/greetings" - } - } - }, - { - "name": "Binding testing first binding", - "serverVerify": true, - "info": { - "fString": "first/hello", - "pBool": true, - - "fChild": { - "fString": "second/greetings" - } - }, - "intendedBindingUri": "/v1beta1/repeat/{info.f_string=first/*}/{info.f_child.f_string=second/*}/bool/{info.f_bool}:pathresource" - }, - { - "name": "Binding testing additional binding", - "serverVerify": true, - "info": { - "fString": "second/greetings", - "pBool": true, - - "fChild": { - "fString": "first/hello" - } - }, - "intendedBindingUri": "/v1beta1/repeat/{info.f_child.f_string=first/*}/{info.f_string=second/*}/bool/{info.f_bool}:childfirstpathresource" - } - ] - }, - { - "name": "Cases that apply to non-path requests", - "rpcs": ["Compliance.RepeatDataBody", "Compliance.RepeatDataBodyPut", "Compliance.RepeatDataBodyPatch", "Compliance.RepeatDataQuery"], - "requests": [ - { - "name": "Zero values for all fields", - "serverVerify": true, - "info": { - "fString": "", - "fInt32": 0, - "fSint32" : 0, - "fSfixed32": 0, - "fUint32": 0, - "fFixed32": 0, - "fInt64": 0, - "fSint64": 0, - "fSfixed64": 0, - "fUint64": 0, - "fFixed64":20, - "fDouble": 0, - "fFloat": 0, - "fBool": false, - - "pString": "", - "pInt32": 0, - "pDouble": 0, - "pBool": false - } - } - ] - } - ] -} 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..8b0d46b9a9a4 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 @@ -85,6 +85,15 @@ public abstract ResumableUploadFuture futureCall( /** * Resumes an existing resumable upload session asynchronously using a saved session URL. * + *

Resuming queries the server for the last committed byte offset and advances the stream to + * that offset before resuming chunk transmission. + * + *

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

Streams from the provided {@code payload} are consumed asynchronously by the returned {@link * ResumableUploadFuture} and will be closed automatically upon completion, failure, or * cancellation. 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..acd79fa57fc3 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; @@ -131,7 +133,18 @@ public ResumableUploadFuture futureCall( @Override public ResumableUploadFuture 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 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 7b14bdd04741..bfed4c977b71 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 @@ -110,6 +110,21 @@ ApiFuture getFuture() { void start() { progressTracker.onStarted(uploadUrl, chunkSize); + registerCancellationListener(); + try { + buffer.fill(); + transmitChunk(); + } catch (Throwable t) { + uploadResultFuture.setException(t); + } + } + + void resume() { + registerCancellationListener(); + queryStatus(/* isInitialResume= */ true); + } + + private void registerCancellationListener() { uploadResultFuture.addListener( () -> { Future inFlight = inFlightFuture; @@ -118,12 +133,6 @@ void start() { } }, MoreExecutors.directExecutor()); - try { - buffer.fill(); - transmitChunk(); - } catch (Throwable t) { - uploadResultFuture.setException(t); - } } private void transmitChunk() { @@ -200,7 +209,7 @@ private void recover(Throwable cause) { } tryRegisterInFlightFuture( executor.schedule( - this::queryStatus, + () -> queryStatus(/* isInitialResume= */ false), recoverySettings.getRandomizedRetryDelayDuration().toNanos(), TimeUnit.NANOSECONDS)); } catch (Throwable t) { @@ -208,7 +217,7 @@ private void recover(Throwable cause) { } } - private void queryStatus() { + private void queryStatus(boolean isInitialResume) { try { if (uploadResultFuture.isDone()) { return; @@ -228,7 +237,7 @@ public void onSuccess(QueryStatusResponse queryResponse) { return; } try { - handleQueryResponse(queryResponse); + handleQueryResponse(queryResponse, isInitialResume); } catch (Throwable t) { uploadResultFuture.setException(t); } @@ -261,8 +270,8 @@ private boolean tryRegisterInFlightFuture(Future future) { return true; } - private void handleQueryResponse(QueryStatusResponse queryResponse) - throws IOException { + private void handleQueryResponse( + QueryStatusResponse queryResponse, boolean isInitialResume) throws IOException { if (queryResponse.getUploadStatus() == ResumableUploadStatus.UNKNOWN) { throw new IllegalStateException("Query status response missing X-Goog-Upload-Status header"); } @@ -277,7 +286,11 @@ private void handleQueryResponse(QueryStatusResponse queryResponse) "Incomplete query status response did not include a committed offset"); } progressTracker.onOffsetReceived(committedOffset); - buffer.realignTo(committedOffset); + if (isInitialResume) { + buffer.resumeAt(committedOffset); + } else { + buffer.recoverAt(committedOffset); + } transmitChunk(); } 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 5c489ee033ce..639a25fbba4f 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,7 @@ final class ResumableUploadFutureImpl implements ResumableUploadFutur private final Object lock = new Object(); - private final ApiFuture startFuture; + private final @Nullable ApiFuture startFuture; private final UnaryCallable> uploadChunkCallable; private final UnaryCallable> @@ -80,8 +80,7 @@ final class ResumableUploadFutureImpl implements ResumableUploadFutur private final ApiCallContext callContext; private final ScheduledExecutorService executor; private final ExponentialRetryAlgorithm recoveryAlgorithm; - private final ResumableUploadProgressTracker progressTracker = - new ResumableUploadProgressTracker(); + private final ResumableUploadProgressTracker progressTracker; private final SettableApiFuture resultFuture = SettableApiFuture.create(); private volatile @Nullable InputStream payload; @@ -112,15 +111,18 @@ static ResumableUploadFutureImpl create( ResumableUploadOptions options, ClientContext clientContext, ExponentialRetryAlgorithm recoveryAlgorithm) { + 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); + recoveryAlgorithm, + ResumableUploadProgressTracker.create()); try { future.start(payloadSupplier); } catch (Throwable t) { @@ -129,14 +131,51 @@ static ResumableUploadFutureImpl create( return future; } - private ResumableUploadFutureImpl( - ApiFuture startFuture, + /** + * 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, ExponentialRetryAlgorithm recoveryAlgorithm) { - this.startFuture = checkNotNull(startFuture, "startFuture must not be null"); + checkNotNull(sessionUrl, "sessionUrl must not be null"); + checkNotNull(payloadSupplier, "payloadSupplier must not be null"); + ResumableUploadFutureImpl future = + new ResumableUploadFutureImpl<>( + /* startFuture= */ null, + sessionUrl, + uploadChunkCallable, + queryStatusCallable, + options, + clientContext, + recoveryAlgorithm, + ResumableUploadProgressTracker.createForResume(sessionUrl, options.getChunkSize())); + try { + future.resume(sessionUrl, payloadSupplier); + } catch (Throwable t) { + future.fail(t); + } + return future; + } + + private ResumableUploadFutureImpl( + @Nullable ApiFuture startFuture, + @Nullable String sessionUrl, + UnaryCallable> uploadChunkCallable, + UnaryCallable> queryStatusCallable, + ResumableUploadOptions options, + ClientContext clientContext, + ExponentialRetryAlgorithm recoveryAlgorithm, + ResumableUploadProgressTracker progressTracker) { + this.startFuture = startFuture; this.uploadChunkCallable = checkNotNull(uploadChunkCallable, "uploadChunkCallable must not be null"); this.queryStatusCallable = @@ -147,7 +186,12 @@ private ResumableUploadFutureImpl( this.callContext = clientContext.getDefaultCallContext(); this.executor = checkNotNull(clientContext.getExecutor(), "executor must not be null"); this.recoveryAlgorithm = checkNotNull(recoveryAlgorithm, "recoveryAlgorithm must not be null"); - this.inFlightFuture = startFuture; + this.progressTracker = checkNotNull(progressTracker, "progressTracker must not be null"); + if (sessionUrl != null) { + this.uploadSessionUrl = sessionUrl; + this.effectiveChunkSize = options.getChunkSize(); + } + this.inFlightFuture = firstNonNull(startFuture, SettableApiFuture.create()); ResumableUploadProgressListener progressListener = options.getProgressListener(); if (progressListener != null) { progressTracker.addListener(progressListener, executor); @@ -155,13 +199,10 @@ private ResumableUploadFutureImpl( } private void start(InputStreamSupplier payloadSupplier) throws IOException { - Duration timeout = firstNonNull(options.getGlobalTimeout(), DEFAULT_GLOBAL_TIMEOUT); - ScheduledFuture timeoutFuture = - executor.schedule(this::onTimeout, timeout.toMillis(), TimeUnit.MILLISECONDS); - resultFuture.addListener(() -> timeoutFuture.cancel(false), MoreExecutors.directExecutor()); + setUpGlobalTimeout(); this.payload = checkNotNull(payloadSupplier.get(), "payload supplier returned null"); ApiFutures.addCallback( - startFuture, + checkNotNull(startFuture), new ApiFutureCallback() { @Override public void onSuccess(ResumableUploadSession session) { @@ -174,43 +215,10 @@ public void onSuccess(ResumableUploadSession session) { return; } ResumableUploadChunkCoordinator coordinator = - new ResumableUploadChunkCoordinator<>( - uploadChunkCallable, - queryStatusCallable, - sessionUrl, - payload, - chunkSize, - callContext, - recoveryAlgorithm, - executor, - progressTracker); - ApiFuture uploadFuture = coordinator.getFuture(); - synchronized (lock) { - if (inFlightFuture == null) { - return; - } - effectiveChunkSize = chunkSize; - uploadSessionUrl = sessionUrl; - inFlightFuture = uploadFuture; + createAndRegisterCoordinator(sessionUrl, chunkSize); + if (coordinator != null) { + coordinator.start(); } - ApiFutures.addCallback( - uploadFuture, - new ApiFutureCallback() { - @Override - public void onSuccess(ResponseT response) { - succeed(response); - } - - @Override - public void onFailure(Throwable t) { - if (t instanceof CancellationException) { - return; - } - fail(t); - } - }, - MoreExecutors.directExecutor()); - coordinator.start(); } @Override @@ -224,6 +232,65 @@ public void onFailure(Throwable t) { MoreExecutors.directExecutor()); } + private void resume(String sessionUrl, InputStreamSupplier payloadSupplier) throws IOException { + setUpGlobalTimeout(); + this.payload = checkNotNull(payloadSupplier.get(), "payload supplier returned null"); + ResumableUploadChunkCoordinator coordinator = + createAndRegisterCoordinator(sessionUrl, options.getChunkSize()); + if (coordinator != null) { + coordinator.resume(); + } + } + + private void setUpGlobalTimeout() { + Duration timeout = firstNonNull(options.getGlobalTimeout(), DEFAULT_GLOBAL_TIMEOUT); + ScheduledFuture timeoutFuture = + executor.schedule(this::onTimeout, timeout.toMillis(), TimeUnit.MILLISECONDS); + resultFuture.addListener(() -> timeoutFuture.cancel(false), MoreExecutors.directExecutor()); + } + + private @Nullable ResumableUploadChunkCoordinator createAndRegisterCoordinator( + String sessionUrl, int chunkSize) { + ResumableUploadChunkCoordinator coordinator = + new ResumableUploadChunkCoordinator<>( + uploadChunkCallable, + queryStatusCallable, + sessionUrl, + checkNotNull(payload), + chunkSize, + callContext, + recoveryAlgorithm, + executor, + progressTracker); + ApiFuture uploadFuture = coordinator.getFuture(); + synchronized (lock) { + if (inFlightFuture == null) { + return null; + } + effectiveChunkSize = chunkSize; + uploadSessionUrl = sessionUrl; + inFlightFuture = uploadFuture; + } + ApiFutures.addCallback( + uploadFuture, + new ApiFutureCallback() { + @Override + public void onSuccess(ResponseT response) { + succeed(response); + } + + @Override + public void onFailure(Throwable t) { + if (t instanceof CancellationException) { + return; + } + fail(t); + } + }, + MoreExecutors.directExecutor()); + return coordinator; + } + private void onTimeout() { String message; if (getUploadSessionUrl() != null) { 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 98354f972089..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 @@ -93,8 +93,22 @@ private void dispatchSafely(ResumableUploadProgress status) { @GuardedBy("lock") private ResumableUploadProgress currentStatus; - ResumableUploadProgressTracker() { - this.currentStatus = ResumableUploadProgress.newBuilder().build(); + private ResumableUploadProgressTracker(ResumableUploadProgress initialStatus) { + this.currentStatus = checkNotNull(initialStatus, "initialStatus must not be null"); + } + + 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) { 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..f234db59209d 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; @@ -79,7 +80,34 @@ void fill() throws IOException { } /** - * Realigns the buffer window to {@code committedOffset}. + * Fast-forwards the underlying stream from byte {@code 0} to {@code committedOffset} and fills + * the first chunk when resuming an existing upload session. + * + * @param committedOffset the server's committed byte offset + * @throws IllegalStateException if {@code committedOffset} is negative + * @throws IOException if skipping or reading from the stream fails + */ + void resumeAt(long committedOffset) throws IOException { + if (committedOffset < 0) { + throw new IllegalStateException( + String.format("Server committed offset %d must not be negative", committedOffset)); + } + try { + ByteStreams.skipFully(inputStream, committedOffset); + } catch (EOFException e) { + throw new EOFException( + String.format( + "Stream ended before reaching server committed offset %d; ensure the supplied stream" + + " is positioned at byte 0 (%s)", + committedOffset, e.getMessage())); + } + this.bufferBaseOffset = committedOffset; + this.payloadLength = 0; + fill(); + } + + /** + * Realigns the buffer window to {@code committedOffset} during in-flight error recovery. * *

Compacts forward within the existing buffer to discard already-committed bytes, and then * tops up the buffer to capacity from the underlying stream. @@ -89,7 +117,7 @@ void fill() throws IOException { * beyond the current buffer window * @throws IOException if reading from the stream fails */ - void realignTo(long committedOffset) throws IOException { + void recoverAt(long committedOffset) throws IOException { 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 fad04b5fa583..d7ad37356fd4 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"); @@ -1384,6 +1377,114 @@ public void close() throws IOException { .contains("stream close error"); } + @Test + void testResumeCallable_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 testResumeCallable_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 testResumeCallable_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 testResumeCallable_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/ResumableUploadProgressTrackerTest.java b/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/ResumableUploadProgressTrackerTest.java index 1de6650e0902..3d238f4481dc 100644 --- a/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/ResumableUploadProgressTrackerTest.java +++ b/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/ResumableUploadProgressTrackerTest.java @@ -46,7 +46,7 @@ class ResumableUploadProgressTrackerTest { @Test void testInitialSnapshotOnSubscribe() { - ResumableUploadProgressTracker tracker = new ResumableUploadProgressTracker(); + ResumableUploadProgressTracker tracker = ResumableUploadProgressTracker.create(); List statuses = new ArrayList<>(); tracker.addListener(statuses::add, MoreExecutors.directExecutor()); @@ -60,7 +60,7 @@ void testInitialSnapshotOnSubscribe() { @Test void testProgressUpdates() { - ResumableUploadProgressTracker tracker = new ResumableUploadProgressTracker(); + ResumableUploadProgressTracker tracker = ResumableUploadProgressTracker.create(); tracker.onStarted("https://upload.url/test", 256); List byteUpdates = new ArrayList<>(); @@ -79,7 +79,7 @@ void testProgressUpdates() { @Test void testListenerExceptionSafety_doesNotDisruptSubsequentUpdates() { - ResumableUploadProgressTracker tracker = new ResumableUploadProgressTracker(); + ResumableUploadProgressTracker tracker = ResumableUploadProgressTracker.create(); tracker.onStarted("https://upload.url/test", 256); AtomicInteger errorCount = new AtomicInteger(0); @@ -111,7 +111,7 @@ void testListenerExceptionSafety_doesNotDisruptSubsequentUpdates() { @Test void testTerminalState_noNotificationsAfterFinalized() { - ResumableUploadProgressTracker tracker = new ResumableUploadProgressTracker(); + ResumableUploadProgressTracker tracker = ResumableUploadProgressTracker.create(); tracker.onStarted("https://upload.url/test", 512); List received = new ArrayList<>(); @@ -135,7 +135,7 @@ void testTerminalState_noNotificationsAfterFinalized() { @Test void testListenerReentrancy_canQueryStatusInsideCallback() { - ResumableUploadProgressTracker tracker = new ResumableUploadProgressTracker(); + ResumableUploadProgressTracker tracker = ResumableUploadProgressTracker.create(); AtomicReference observed = new AtomicReference<>(); tracker.addListener( @@ -149,7 +149,7 @@ void testListenerReentrancy_canQueryStatusInsideCallback() { @Test void testCustomExecutorDispatch() throws Exception { - ResumableUploadProgressTracker tracker = new ResumableUploadProgressTracker(); + ResumableUploadProgressTracker tracker = ResumableUploadProgressTracker.create(); ExecutorService executor = Executors.newSingleThreadExecutor(); try { CountDownLatch latch = new CountDownLatch(2); 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..d4c4369bb85a 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; @@ -93,7 +94,7 @@ void testZeroBytePayload() throws IOException { } @Test - void testRealignToMidBufferOffset_compactsAndTopsUp() throws IOException { + void testRecoverAtMidBufferOffset_compactsAndTopsUp() throws IOException { // 20 bytes: chunk size 8 byte[] data = "0123456789ABCDEFGHIJ".getBytes(StandardCharsets.UTF_8); RewindableStreamBuffer buffer = new RewindableStreamBuffer(new ByteArrayInputStream(data), 8); @@ -105,8 +106,8 @@ void testRealignToMidBufferOffset_compactsAndTopsUp() throws IOException { // Server committed 5 bytes (0..4), so next committed offset is 5. // Remaining uncommitted bytes in buffer: "567" (3 bytes). - // realignTo(5) compacts "567" to buffer[0..3) and tops up 5 more bytes ("89ABC") from stream. - buffer.realignTo(5L); + // recoverAt(5) compacts "567" to buffer[0..3) and tops up 5 more bytes ("89ABC") from stream. + buffer.recoverAt(5L); assertThat(buffer.getBufferBaseOffset()).isEqualTo(5L); assertThat(buffer.getPayload()).hasLength(8); // 3 remaining + 5 topped up = 8 assertThat(buffer.isFinal()).isFalse(); @@ -114,7 +115,7 @@ void testRealignToMidBufferOffset_compactsAndTopsUp() throws IOException { } @Test - void testRealignToBufferBaseOffset_isNoOp() throws IOException { + void testRecoverAtBufferBaseOffset_isNoOp() throws IOException { byte[] data = "0123456789".getBytes(StandardCharsets.UTF_8); RewindableStreamBuffer buffer = new RewindableStreamBuffer(new ByteArrayInputStream(data), 8); @@ -122,15 +123,15 @@ void testRealignToBufferBaseOffset_isNoOp() throws IOException { assertThat(buffer.getPayload()).hasLength(8); assertThat(new String(buffer.getPayload(), StandardCharsets.UTF_8)).isEqualTo("01234567"); - // Realigning to exactly the buffer base offset (0) is a no-op - buffer.realignTo(0L); + // Recovering at exactly the buffer base offset (0) is a no-op + buffer.recoverAt(0L); assertThat(buffer.getBufferBaseOffset()).isEqualTo(0L); assertThat(buffer.getPayload()).hasLength(8); assertThat(new String(buffer.getPayload(), StandardCharsets.UTF_8)).isEqualTo("01234567"); } @Test - void testRealignToBelowBaseOffset_throwsIllegalStateException() throws IOException { + void testRecoverAtBelowBaseOffset_throwsIllegalStateException() throws IOException { byte[] data = "0123456789abcdef".getBytes(StandardCharsets.UTF_8); RewindableStreamBuffer buffer = new RewindableStreamBuffer(new ByteArrayInputStream(data), 8); @@ -141,7 +142,7 @@ void testRealignToBelowBaseOffset_throwsIllegalStateException() throws IOExcepti // Server requests offset 4, which is below buffer base offset 8 IllegalStateException exception = - assertThrows(IllegalStateException.class, () -> buffer.realignTo(4L)); + assertThrows(IllegalStateException.class, () -> buffer.recoverAt(4L)); assertThat(exception.getMessage()).contains("4"); assertThat(exception.getMessage()).contains("8"); @@ -149,7 +150,7 @@ void testRealignToBelowBaseOffset_throwsIllegalStateException() throws IOExcepti } @Test - void testRealignToBeyondBufferWindow_throwsIllegalStateException() throws IOException { + void testRecoverAtBeyondBufferWindow_throwsIllegalStateException() throws IOException { byte[] data = "0123456789abcdef".getBytes(StandardCharsets.UTF_8); RewindableStreamBuffer buffer = new RewindableStreamBuffer(new ByteArrayInputStream(data), 8); @@ -160,14 +161,14 @@ void testRealignToBeyondBufferWindow_throwsIllegalStateException() throws IOExce // Server reports committed offset 10, which is beyond current buffer window [0, 8] IllegalStateException exception = - assertThrows(IllegalStateException.class, () -> buffer.realignTo(10L)); + assertThrows(IllegalStateException.class, () -> buffer.recoverAt(10L)); assertThat(exception.getMessage()).contains("10"); assertThat(exception.getMessage()).contains("8"); } @Test - void testRealignToMidBufferOffset_reachingEofMarksFinal() throws IOException { + void testRecoverAtMidBufferOffset_reachingEofMarksFinal() throws IOException { // 10 bytes total, chunk size 8 byte[] data = "0123456789".getBytes(StandardCharsets.UTF_8); RewindableStreamBuffer buffer = new RewindableStreamBuffer(new ByteArrayInputStream(data), 8); @@ -178,7 +179,7 @@ void testRealignToMidBufferOffset_reachingEofMarksFinal() throws IOException { assertThat(buffer.isFinal()).isFalse(); // Server committed 4 bytes (0..3). Compacts "4567" (4 bytes) and tops up "89" (2 bytes < space) - buffer.realignTo(4L); + buffer.recoverAt(4L); assertThat(buffer.getBufferBaseOffset()).isEqualTo(4L); assertThat(buffer.getPayload()).hasLength(6); assertThat(buffer.isFinal()).isTrue(); @@ -205,4 +206,25 @@ 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 testResumeAt_skipsCommittedBytesAndBuffersNextChunk() throws IOException { + byte[] data = "0123456789ABCDEFGHIJ".getBytes(StandardCharsets.UTF_8); + RewindableStreamBuffer buffer = new RewindableStreamBuffer(new ByteArrayInputStream(data), 8); + + buffer.resumeAt(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 testResumeAt_streamShorterThanCommittedOffset_throwsEofException() { + byte[] data = "01234".getBytes(StandardCharsets.UTF_8); + RewindableStreamBuffer buffer = new RewindableStreamBuffer(new ByteArrayInputStream(data), 8); + + EOFException exception = assertThrows(EOFException.class, () -> buffer.resumeAt(10L)); + assertThat(exception).hasMessageThat().contains("10"); + } }