From b02ed8c6e1ec1db1eb9c7fc94027f13b7ba13808 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 | 10 ++ .../gax/rpc/ResumableUploadCallableImpl.java | 15 +- .../rpc/ResumableUploadChunkCoordinator.java | 11 +- .../gax/rpc/ResumableUploadFutureImpl.java | 63 +++++++- .../rpc/ResumableUploadProgressTracker.java | 10 ++ .../api/gax/rpc/RewindableStreamBuffer.java | 37 ++++- .../rpc/ResumableUploadCallableImplTest.java | 141 +++++++++++++++++- .../gax/rpc/RewindableStreamBufferTest.java | 34 +++++ 11 files changed, 366 insertions(+), 21 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..39e38efefdb8 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,16 @@ 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 skips the stream forward + * to that offset before resuming chunk transmission. {@code payload} 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 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 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 b434d73e5a43..c356847c3c66 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> @@ -111,10 +111,12 @@ 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, @@ -129,15 +131,51 @@ static ResumableUploadFutureImpl create( 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, + ExponentialRetryAlgorithm recoveryAlgorithm) { + 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( - ApiFuture startFuture, + @Nullable 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"); + this.startFuture = startFuture; this.uploadChunkCallable = checkNotNull(uploadChunkCallable, "uploadChunkCallable must not be null"); this.queryStatusCallable = @@ -149,7 +187,12 @@ private ResumableUploadFutureImpl( 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; + if (sessionUrl != null) { + this.uploadSessionUrl = sessionUrl; + this.effectiveChunkSize = options.getChunkSize(); + } + // Non-null placeholder marks the upload as active until the coordinator registers. + this.inFlightFuture = firstNonNull(startFuture, SettableApiFuture.create()); ResumableUploadProgressListener progressListener = options.getProgressListener(); if (progressListener != null) { progressTracker.addListener(progressListener, executor); @@ -160,7 +203,7 @@ private void start(InputStreamSupplier payloadSupplier) throws IOException { setUpGlobalTimeout(); this.payload = checkNotNull(payloadSupplier.get(), "payload supplier returned null"); ApiFutures.addCallback( - startFuture, + checkNotNull(startFuture), new ApiFutureCallback() { @Override public void onSuccess(ResumableUploadSession session) { @@ -190,6 +233,16 @@ 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 = 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"); + } }