diff --git a/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITAppendableUploadFakeTest.java b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITAppendableUploadFakeTest.java index 78d07c4d9425..dcf66712a553 100644 --- a/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITAppendableUploadFakeTest.java +++ b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITAppendableUploadFakeTest.java @@ -291,6 +291,216 @@ public void bidiWriteObjectRedirectedError_maxAttempts() throws Exception { } } + @Test + public void bidiWriteObjectRedirectedError_redirectCounterResetOnResponse() throws Exception { + String routingToken1 = "routingToken1"; + String routingToken2 = "routingToken2"; + String routingToken3 = "routingToken3"; + String routingToken4 = "routingToken4"; + String routingToken5 = "routingToken5"; + + BidiWriteHandle writeHandle1 = + BidiWriteHandle.newBuilder().setHandle(ByteString.copyFromUtf8("handle-1")).build(); + BidiWriteHandle writeHandle2 = + BidiWriteHandle.newBuilder().setHandle(ByteString.copyFromUtf8("handle-2")).build(); + BidiWriteHandle writeHandle3 = + BidiWriteHandle.newBuilder().setHandle(ByteString.copyFromUtf8("handle-3")).build(); + + // 1. First write of "ABC". + BidiWriteObjectRequest req_open_abc = BidiUploadTestUtils.withFlushAndStateLookup(open_abc); + BidiWriteObjectResponse res_abc_h1 = res_abc.toBuilder().setWriteHandle(writeHandle1).build(); + + // 2. Second write of "DEF". + BidiWriteObjectRequest req_def = BidiUploadTestUtils.withFlushAndStateLookup(def); + BidiWriteObjectResponse res_def_success = + BidiWriteObjectResponse.newBuilder() + .setWriteHandle(writeHandle1) + .setPersistedSize(6) + .build(); + + // 3. Reconnect to routingToken1 + BidiWriteObjectRequest reconnect_token1 = + BidiWriteObjectRequest.newBuilder() + .setAppendObjectSpec( + AppendObjectSpec.newBuilder() + .setBucket(METADATA.getBucket()) + .setObject(METADATA.getName()) + .setGeneration(METADATA.getGeneration()) + .setWriteHandle(writeHandle1) + .setRoutingToken(routingToken1) + .build()) + .setStateLookup(true) + .build(); + BidiWriteObjectResponse res_lookup_token1 = + BidiWriteObjectResponse.newBuilder() + .setWriteHandle(writeHandle1) + .setPersistedSize(3) + .build(); + + // 4. Third write of "GHI". + BidiWriteObjectRequest req_ghi = BidiUploadTestUtils.withFlushAndStateLookup(ghi); + BidiWriteObjectResponse res_ghi_success = + BidiWriteObjectResponse.newBuilder() + .setWriteHandle(writeHandle2) + .setPersistedSize(9) + .build(); + + // 5. Reconnect to routingToken2 + BidiWriteObjectRequest reconnect_token2 = + BidiWriteObjectRequest.newBuilder() + .setAppendObjectSpec( + AppendObjectSpec.newBuilder() + .setBucket(METADATA.getBucket()) + .setObject(METADATA.getName()) + .setGeneration(METADATA.getGeneration()) + .setWriteHandle(writeHandle1) + .setRoutingToken(routingToken2) + .build()) + .setStateLookup(true) + .build(); + BidiWriteObjectResponse res_lookup_token2 = + BidiWriteObjectResponse.newBuilder() + .setWriteHandle(writeHandle2) + .setPersistedSize(6) + .build(); + + // 6. Fourth write of "J" (finalize/finish write) + BidiWriteObjectRequest req_j_finish = j_finish; + BidiWriteObjectResponse res_j_final = + resource_10.toBuilder().setWriteHandle(writeHandle3).build(); + + // 7. Reconnect to routingToken3 + BidiWriteObjectRequest reconnect_token3 = + BidiWriteObjectRequest.newBuilder() + .setAppendObjectSpec( + AppendObjectSpec.newBuilder() + .setBucket(METADATA.getBucket()) + .setObject(METADATA.getName()) + .setGeneration(METADATA.getGeneration()) + .setWriteHandle(writeHandle2) + .setRoutingToken(routingToken3) + .build()) + .setStateLookup(true) + .build(); + + // 8. Reconnect to routingToken4 + BidiWriteObjectRequest reconnect_token4 = + BidiWriteObjectRequest.newBuilder() + .setAppendObjectSpec( + AppendObjectSpec.newBuilder() + .setBucket(METADATA.getBucket()) + .setObject(METADATA.getName()) + .setGeneration(METADATA.getGeneration()) + .setWriteHandle(writeHandle2) + .setRoutingToken(routingToken4) + .build()) + .setStateLookup(true) + .build(); + + // 9. Reconnect to routingToken5 + BidiWriteObjectRequest reconnect_token5 = + BidiWriteObjectRequest.newBuilder() + .setAppendObjectSpec( + AppendObjectSpec.newBuilder() + .setBucket(METADATA.getBucket()) + .setObject(METADATA.getName()) + .setGeneration(METADATA.getGeneration()) + .setWriteHandle(writeHandle2) + .setRoutingToken(routingToken5) + .build()) + .setStateLookup(true) + .build(); + BidiWriteObjectResponse res_lookup_token5 = + BidiWriteObjectResponse.newBuilder() + .setWriteHandle(writeHandle3) + .setPersistedSize(9) + .build(); + + AtomicInteger defCounter = new AtomicInteger(); + Consumer> defHandler = + respond -> { + if (defCounter.getAndIncrement() == 0) { + respond.onError(packRedirectIntoAbortedException(makeRedirect(routingToken1))); + } else { + respond.onNext(res_def_success); + } + }; + + AtomicInteger ghiCounter = new AtomicInteger(); + Consumer> ghiHandler = + respond -> { + if (ghiCounter.getAndIncrement() == 0) { + respond.onError(packRedirectIntoAbortedException(makeRedirect(routingToken2))); + } else { + respond.onNext(res_ghi_success); + } + }; + + AtomicInteger jCounter = new AtomicInteger(); + Consumer> jHandler = + respond -> { + if (jCounter.getAndIncrement() == 0) { + respond.onError(packRedirectIntoAbortedException(makeRedirect(routingToken3))); + } else { + respond.onNext(res_j_final); + respond.onCompleted(); + } + }; + + FakeStorage fake = + FakeStorage.of( + ImmutableMap + .>> + builder() + .put(req_open_abc, respond -> respond.onNext(res_abc_h1)) + .put(req_def, defHandler) + .put(reconnect_token1, respond -> respond.onNext(res_lookup_token1)) + .put(req_ghi, ghiHandler) + .put(reconnect_token2, respond -> respond.onNext(res_lookup_token2)) + .put(req_j_finish, jHandler) + .put( + reconnect_token3, + respond -> + respond.onError( + packRedirectIntoAbortedException(makeRedirect(routingToken4)))) + .put( + reconnect_token4, + respond -> + respond.onError( + packRedirectIntoAbortedException(makeRedirect(routingToken5)))) + .put(reconnect_token5, respond -> respond.onNext(res_lookup_token5)) + .build()); + + try (FakeServer fakeServer = FakeServer.of(fake); + Storage storage = + fakeServer.getGrpcStorageOptions().toBuilder() + .setRetrySettings( + fakeServer.getGrpcStorageOptions().getRetrySettings().toBuilder() + .setRetryDelayMultiplier(1.0) + .setInitialRetryDelayDuration(Duration.ofMillis(10)) + .build()) + .build() + .getService()) { + + BlobId id = BlobId.of("b", "o"); + BlobAppendableUploadConfig config = + BlobAppendableUploadConfig.of() + .withFlushPolicy(FlushPolicy.maxFlushSize(3)) + .withCloseAction(CloseAction.FINALIZE_WHEN_CLOSING); + BlobAppendableUpload b = + storage.blobAppendableUpload(BlobInfo.newBuilder(id).build(), config); + try (AppendableUploadWriteableByteChannel channel = b.open()) { + ByteBuffer wrap = ByteBuffer.wrap(content.getBytes()); + Buffers.emptyTo(wrap, channel); + } + + // Verification + ApiFuture resultFuture = b.getResult(); + BlobInfo finalMetadata = resultFuture.get(3, TimeUnit.SECONDS); + assertThat(finalMetadata.getSize()).isEqualTo(10L); + } + } + /** * We use a small segmenter (3 byte segments) and flush "ABCDEFGHIJ". We make sure that this * resolves to segments of "ABC"/"DEF"/"GHI"/"J". diff --git a/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITAppendableUploadTest.java b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITAppendableUploadTest.java index c5e9b605857f..94192e2350d8 100644 --- a/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITAppendableUploadTest.java +++ b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITAppendableUploadTest.java @@ -33,14 +33,12 @@ import com.google.cloud.storage.it.ChecksummedTestContent; import com.google.cloud.storage.it.runner.StorageITRunner; import com.google.cloud.storage.it.runner.annotations.Backend; -import com.google.cloud.storage.it.runner.annotations.BucketFixture; -import com.google.cloud.storage.it.runner.annotations.BucketType; import com.google.cloud.storage.it.runner.annotations.CrossRun; import com.google.cloud.storage.it.runner.annotations.Inject; +import com.google.cloud.storage.it.runner.annotations.LocationType; import com.google.cloud.storage.it.runner.annotations.Parameterized; import com.google.cloud.storage.it.runner.annotations.Parameterized.Parameter; import com.google.cloud.storage.it.runner.annotations.Parameterized.ParametersProvider; -import com.google.cloud.storage.it.runner.registry.Generator; import com.google.common.base.MoreObjects; import com.google.common.collect.ImmutableList; import com.google.common.io.ByteStreams; @@ -59,17 +57,16 @@ @RunWith(StorageITRunner.class) @CrossRun( backends = {Backend.TEST_BENCH, Backend.PROD}, - transports = Transport.GRPC) + transports = Transport.GRPC, + locations = {LocationType.ZONAL_RAPID, LocationType.REGIONAL_RAPID}) +// TODO: Enable REGIONAL_RAPID tests once RCU is available in prod +@CrossRun.Ignore(locations = {LocationType.REGIONAL_RAPID}) @Parameterized(UploadConfigParameters.class) public final class ITAppendableUploadTest { - @Inject public Generator generator; - @Inject public Storage storage; - @Inject - @BucketFixture(BucketType.RAPID) - public BucketInfo bucket; + @Inject public BucketInfo bucket; @Inject public Backend backend; @@ -86,7 +83,10 @@ public void appendableUpload_emptyObject() BlobAppendableUpload upload = storage.blobAppendableUpload( - BlobInfo.newBuilder(bucket, UUID.randomUUID().toString()).build(), p.uploadConfig); + BlobInfo.newBuilder(bucket, UUID.randomUUID().toString()) + .setStorageClass(StorageClass.valueOf("RAPID")) + .build(), + p.uploadConfig); upload.open().close(); @@ -109,7 +109,10 @@ public void appendableUpload_bytes() BlobAppendableUpload upload = storage.blobAppendableUpload( - BlobInfo.newBuilder(bucket, UUID.randomUUID().toString()).build(), p.uploadConfig); + BlobInfo.newBuilder(bucket, UUID.randomUUID().toString()) + .setStorageClass(StorageClass.valueOf("RAPID")) + .build(), + p.uploadConfig); // cut out the middle + 1 byte int length = p.content.length(); @@ -141,7 +144,10 @@ public void explicitFlush() BlobAppendableUpload upload = storage.blobAppendableUpload( - BlobInfo.newBuilder(bucket, UUID.randomUUID().toString()).build(), p.uploadConfig); + BlobInfo.newBuilder(bucket, UUID.randomUUID().toString()) + .setStorageClass(StorageClass.valueOf("RAPID")) + .build(), + p.uploadConfig); try (AppendableUploadWriteableByteChannel channel = upload.open()) { ByteBuffer src = p.content.asByteBuffer(); @@ -179,7 +185,9 @@ public void appendableBlobUploadTakeover() throws Exception { p.uploadConfig.withCloseAction(CloseAction.CLOSE_WITHOUT_FINALIZING); BlobAppendableUpload upload = - storage.blobAppendableUpload(BlobInfo.newBuilder(id).build(), doNotFinalizeConfig); + storage.blobAppendableUpload( + BlobInfo.newBuilder(id).setStorageClass(StorageClass.valueOf("RAPID")).build(), + doNotFinalizeConfig); try (AppendableUploadWriteableByteChannel channel = upload.open()) { int written = Buffers.emptyTo(ByteBuffer.wrap(c1.getBytes()), channel); assertThat(written).isEqualTo(c1.length()); @@ -190,7 +198,10 @@ public void appendableBlobUploadTakeover() throws Exception { BlobAppendableUpload takeOver = storage.blobAppendableUpload( - BlobInfo.newBuilder(done1.getBlobId()).build(), p.uploadConfig); + BlobInfo.newBuilder(done1.getBlobId()) + .setStorageClass(StorageClass.valueOf("RAPID")) + .build(), + p.uploadConfig); try (AppendableUploadWriteableByteChannel channel = takeOver.open()) { int written = Buffers.emptyTo(ByteBuffer.wrap(c2.getBytes()), channel); assertThat(written).isEqualTo(c2.length()); @@ -219,7 +230,9 @@ public void testUploadFileUsingAppendable() throws Exception { } BlobAppendableUpload appendable = - storage.blobAppendableUpload(BlobInfo.newBuilder(bid).build(), p.uploadConfig); + storage.blobAppendableUpload( + BlobInfo.newBuilder(bid).setStorageClass(StorageClass.valueOf("RAPID")).build(), + p.uploadConfig); try (SeekableByteChannel r = tmpFile.reader(); AppendableUploadWriteableByteChannel w = appendable.open()) { long copied = ByteStreams.copy(r, w); @@ -241,7 +254,9 @@ public void takeoverJustToFinalizeWorks() throws Exception { p.uploadConfig.getCloseAction() != CloseAction.FINALIZE_WHEN_CLOSING); BlobAppendableUpload upload = - storage.blobAppendableUpload(BlobInfo.newBuilder(bid).build(), p.uploadConfig); + storage.blobAppendableUpload( + BlobInfo.newBuilder(bid).setStorageClass(StorageClass.valueOf("RAPID")).build(), + p.uploadConfig); try (AppendableUploadWriteableByteChannel channel = upload.open()) { int written = Buffers.emptyTo(ByteBuffer.wrap(p.content.getBytes()), channel); assertThat(written).isEqualTo(p.content.length()); @@ -252,7 +267,10 @@ public void takeoverJustToFinalizeWorks() throws Exception { BlobAppendableUpload takeOver = storage.blobAppendableUpload( - BlobInfo.newBuilder(done1.getBlobId()).build(), p.uploadConfig); + BlobInfo.newBuilder(done1.getBlobId()) + .setStorageClass(StorageClass.valueOf("RAPID")) + .build(), + p.uploadConfig); takeOver.open().finalizeAndClose(); BlobInfo done2 = takeOver.getResult().get(5, TimeUnit.SECONDS); @@ -271,7 +289,9 @@ public void explicitFinalizeWithCorrectChecksum() throws Exception { p.uploadConfig.getCloseAction() != CloseAction.FINALIZE_WHEN_CLOSING); BlobAppendableUpload upload = - storage.blobAppendableUpload(BlobInfo.newBuilder(bid).build(), p.uploadConfig); + storage.blobAppendableUpload( + BlobInfo.newBuilder(bid).setStorageClass(StorageClass.valueOf("RAPID")).build(), + p.uploadConfig); try (AppendableUploadWriteableByteChannel channel = upload.open()) { int written = Buffers.emptyTo(ByteBuffer.wrap(p.content.getBytes()), channel); @@ -295,7 +315,9 @@ public void explicitFinalizeWithIncorrectChecksumFails() throws Exception { p.uploadConfig.getCloseAction() != CloseAction.FINALIZE_WHEN_CLOSING); BlobAppendableUpload upload = - storage.blobAppendableUpload(BlobInfo.newBuilder(bid).build(), p.uploadConfig); + storage.blobAppendableUpload( + BlobInfo.newBuilder(bid).setStorageClass(StorageClass.valueOf("RAPID")).build(), + p.uploadConfig); try (AppendableUploadWriteableByteChannel channel = upload.open()) { int written = Buffers.emptyTo(ByteBuffer.wrap(p.content.getBytes()), channel); @@ -328,7 +350,9 @@ public void takeoverJustToFinalizeWithIncorrectChecksumFails() throws Exception p.uploadConfig.getCloseAction() != CloseAction.FINALIZE_WHEN_CLOSING); BlobAppendableUpload upload = - storage.blobAppendableUpload(BlobInfo.newBuilder(bid).build(), p.uploadConfig); + storage.blobAppendableUpload( + BlobInfo.newBuilder(bid).setStorageClass(StorageClass.valueOf("RAPID")).build(), + p.uploadConfig); try (AppendableUploadWriteableByteChannel channel = upload.open()) { int written = Buffers.emptyTo(ByteBuffer.wrap(p.content.getBytes()), channel); assertThat(written).isEqualTo(p.content.length()); @@ -339,7 +363,10 @@ public void takeoverJustToFinalizeWithIncorrectChecksumFails() throws Exception BlobAppendableUpload takeOver = storage.blobAppendableUpload( - BlobInfo.newBuilder(done1.getBlobId()).build(), p.uploadConfig); + BlobInfo.newBuilder(done1.getBlobId()) + .setStorageClass(StorageClass.valueOf("RAPID")) + .build(), + p.uploadConfig); String badCrc = Utils.crc32cCodec.encode(Crc32cValue.zero().getValue()); try (AppendableUploadWriteableByteChannel channel = takeOver.open()) { @@ -369,7 +396,9 @@ public void takeoverAndAppendWithCorrectChecksumWorks() throws Exception { p.uploadConfig.getCloseAction() != CloseAction.FINALIZE_WHEN_CLOSING); BlobAppendableUpload upload = - storage.blobAppendableUpload(BlobInfo.newBuilder(bid).build(), p.uploadConfig); + storage.blobAppendableUpload( + BlobInfo.newBuilder(bid).setStorageClass(StorageClass.valueOf("RAPID")).build(), + p.uploadConfig); try (AppendableUploadWriteableByteChannel channel = upload.open()) { int written = Buffers.emptyTo(ByteBuffer.wrap(p.content.getBytes()), channel); assertThat(written).isEqualTo(p.content.length()); @@ -379,7 +408,10 @@ public void takeoverAndAppendWithCorrectChecksumWorks() throws Exception { BlobAppendableUpload takeOver = storage.blobAppendableUpload( - BlobInfo.newBuilder(done1.getBlobId()).build(), p.uploadConfig); + BlobInfo.newBuilder(done1.getBlobId()) + .setStorageClass(StorageClass.valueOf("RAPID")) + .build(), + p.uploadConfig); try (AppendableUploadWriteableByteChannel channel = takeOver.open()) { int written = Buffers.emptyTo(ByteBuffer.wrap(p.content.getBytes()), channel); @@ -405,7 +437,9 @@ public void takeoverAndAppendWithIncorrectChecksumFails() throws Exception { p.uploadConfig.getCloseAction() != CloseAction.FINALIZE_WHEN_CLOSING); BlobAppendableUpload upload = - storage.blobAppendableUpload(BlobInfo.newBuilder(bid).build(), p.uploadConfig); + storage.blobAppendableUpload( + BlobInfo.newBuilder(bid).setStorageClass(StorageClass.valueOf("RAPID")).build(), + p.uploadConfig); try (AppendableUploadWriteableByteChannel channel = upload.open()) { int written = Buffers.emptyTo(ByteBuffer.wrap(p.content.getBytes()), channel); assertThat(written).isEqualTo(p.content.length()); @@ -415,7 +449,10 @@ public void takeoverAndAppendWithIncorrectChecksumFails() throws Exception { BlobAppendableUpload takeOver = storage.blobAppendableUpload( - BlobInfo.newBuilder(done1.getBlobId()).build(), p.uploadConfig); + BlobInfo.newBuilder(done1.getBlobId()) + .setStorageClass(StorageClass.valueOf("RAPID")) + .build(), + p.uploadConfig); try (AppendableUploadWriteableByteChannel channel = takeOver.open()) { int written = Buffers.emptyTo(ByteBuffer.wrap(p.content.getBytes()), channel); diff --git a/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITObjectReadSessionFakeTest.java b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITObjectReadSessionFakeTest.java index 1cb32704982b..509db8939119 100644 --- a/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITObjectReadSessionFakeTest.java +++ b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITObjectReadSessionFakeTest.java @@ -210,8 +210,16 @@ public void bidiReadObjectRedirectedError() throws Exception { req3, respond -> respond.onNext(res2))); + GrpcRequestAuditing requestAuditing = new GrpcRequestAuditing(); try (FakeServer fakeServer = FakeServer.of(fake); - Storage storage = fakeServer.getGrpcStorageOptions().toBuilder().build().getService()) { + Storage storage = + fakeServer.getGrpcStorageOptions().toBuilder() + .setGrpcInterceptorProvider( + () -> + ImmutableList.of( + requestAuditing, GrpcPlainRequestLoggingInterceptor.getInstance())) + .build() + .getService()) { BlobId id = BlobId.of("b", "o"); ApiFuture futureBlobDescriptor = storage.blobReadSession(id); @@ -222,6 +230,12 @@ public void bidiReadObjectRedirectedError() throws Exception { .get(1, TimeUnit.SECONDS); assertThat(xxd(actual)).isEqualTo(xxd(content.getBytes())); + + requestAuditing + .assertRequestHeader("x-goog-request-params") + .containsExactly( + "bucket=projects/_/buckets/b", + "bucket=projects/_/buckets/b&routing_token=" + routingToken); } } } @@ -1805,6 +1819,195 @@ static ObjectReadSessionImpl getObjectReadSessionImpl(BlobReadSession bd) { return orsi; } + private static Consumer> onRedirect( + BidiReadHandle handle, String token) { + return respond -> { + BidiReadObjectRedirectedError redirect = + BidiReadObjectRedirectedError.newBuilder() + .setReadHandle(handle) + .setRoutingToken(token) + .build(); + + com.google.rpc.Status grpcStatusDetails = + com.google.rpc.Status.newBuilder() + .setCode(com.google.rpc.Code.UNAVAILABLE_VALUE) + .setMessage("redirect") + .addDetails(Any.pack(redirect)) + .build(); + + Metadata trailers = new Metadata(); + trailers.put(GRPC_STATUS_DETAILS_KEY, grpcStatusDetails); + StatusRuntimeException statusRuntimeException = + Status.UNAVAILABLE.withDescription("redirect").asRuntimeException(trailers); + respond.onError(statusRuntimeException); + }; + } + + @Test + public void bidiReadObjectRedirectedError_redirectCounterResetOnResponse() throws Exception { + BidiReadHandle handle1 = + BidiReadHandle.newBuilder().setHandle(ByteString.copyFromUtf8("handle-1")).build(); + BidiReadHandle handle2 = + BidiReadHandle.newBuilder().setHandle(ByteString.copyFromUtf8("handle-2")).build(); + BidiReadHandle handle3 = + BidiReadHandle.newBuilder().setHandle(ByteString.copyFromUtf8("handle-3")).build(); + BidiReadHandle handle4 = + BidiReadHandle.newBuilder().setHandle(ByteString.copyFromUtf8("handle-4")).build(); + BidiReadHandle handle5 = + BidiReadHandle.newBuilder().setHandle(ByteString.copyFromUtf8("handle-5")).build(); + + BidiReadObjectRequest req_read_1 = read(1, 10, 5); + + BidiReadObjectRequest req_read_1_redirected_1 = + BidiReadObjectRequest.newBuilder() + .setReadObjectSpec( + BidiReadObjectSpec.newBuilder() + .setBucket(METADATA.getBucket()) + .setObject(METADATA.getName()) + .setGeneration(1) + .setReadHandle(handle1) + .setRoutingToken("token-1") + .build()) + .addReadRanges(getReadRange(1, 10, 5)) + .build(); + + BidiReadObjectRequest req_read_2 = + BidiReadObjectRequest.newBuilder().addReadRanges(getReadRange(2, 15, 5)).build(); + + BidiReadObjectRequest req_read_2_redirected_2 = + BidiReadObjectRequest.newBuilder() + .setReadObjectSpec( + BidiReadObjectSpec.newBuilder() + .setBucket(METADATA.getBucket()) + .setObject(METADATA.getName()) + .setGeneration(1) + .setReadHandle(handle2) + .setRoutingToken("token-2") + .build()) + .addReadRanges(getReadRange(2, 15, 5)) + .build(); + + BidiReadObjectRequest req_read_3 = + BidiReadObjectRequest.newBuilder().addReadRanges(getReadRange(3, 20, 5)).build(); + + BidiReadObjectRequest req_read_3_redirected_3 = + BidiReadObjectRequest.newBuilder() + .setReadObjectSpec( + BidiReadObjectSpec.newBuilder() + .setBucket(METADATA.getBucket()) + .setObject(METADATA.getName()) + .setGeneration(1) + .setReadHandle(handle3) + .setRoutingToken("token-3") + .build()) + .addReadRanges(getReadRange(3, 20, 5)) + .build(); + + BidiReadObjectRequest req_read_3_redirected_4 = + BidiReadObjectRequest.newBuilder() + .setReadObjectSpec( + BidiReadObjectSpec.newBuilder() + .setBucket(METADATA.getBucket()) + .setObject(METADATA.getName()) + .setGeneration(1) + .setReadHandle(handle4) + .setRoutingToken("token-4") + .build()) + .addReadRanges(getReadRange(3, 20, 5)) + .build(); + + BidiReadObjectRequest req_read_3_redirected_5 = + BidiReadObjectRequest.newBuilder() + .setReadObjectSpec( + BidiReadObjectSpec.newBuilder() + .setBucket(METADATA.getBucket()) + .setObject(METADATA.getName()) + .setGeneration(1) + .setReadHandle(handle5) + .setRoutingToken("token-5") + .build()) + .addReadRanges(getReadRange(3, 20, 5)) + .build(); + + ChecksummedTestContent content1 = ChecksummedTestContent.of(ALL_OBJECT_BYTES, 10, 5); + BidiReadObjectResponse res_read_1 = + BidiReadObjectResponse.newBuilder() + .setMetadata(METADATA) + .addObjectDataRanges( + ObjectRangeData.newBuilder() + .setChecksummedData(content1.asChecksummedData()) + .setReadRange(getReadRange(1, 10, 5)) + .setRangeEnd(true) + .build()) + .build(); + + ChecksummedTestContent content2 = ChecksummedTestContent.of(ALL_OBJECT_BYTES, 15, 5); + BidiReadObjectResponse res_read_2 = + BidiReadObjectResponse.newBuilder() + .setMetadata(METADATA) + .addObjectDataRanges( + ObjectRangeData.newBuilder() + .setChecksummedData(content2.asChecksummedData()) + .setReadRange(getReadRange(2, 15, 5)) + .setRangeEnd(true) + .build()) + .build(); + + ChecksummedTestContent content3 = ChecksummedTestContent.of(ALL_OBJECT_BYTES, 20, 5); + BidiReadObjectResponse res_read_3 = + BidiReadObjectResponse.newBuilder() + .setMetadata(METADATA) + .addObjectDataRanges( + ObjectRangeData.newBuilder() + .setChecksummedData(content3.asChecksummedData()) + .setReadRange(getReadRange(3, 20, 5)) + .setRangeEnd(true) + .build()) + .build(); + + FakeStorage fake = + FakeStorage.of( + ImmutableMap + .>>builder() + .put(REQ_OPEN, respond -> respond.onNext(RES_OPEN)) + .put(req_read_1, onRedirect(handle1, "token-1")) + .put(req_read_1_redirected_1, respond -> respond.onNext(res_read_1)) + .put(req_read_2, onRedirect(handle2, "token-2")) + .put(req_read_2_redirected_2, respond -> respond.onNext(res_read_2)) + .put(req_read_3, onRedirect(handle3, "token-3")) + .put(req_read_3_redirected_3, onRedirect(handle4, "token-4")) + .put(req_read_3_redirected_4, onRedirect(handle5, "token-5")) + .put(req_read_3_redirected_5, respond -> respond.onNext(res_read_3)) + .build()); + + try (FakeServer fakeServer = FakeServer.of(fake); + Storage storage = fakeServer.getGrpcStorageOptions().toBuilder().build().getService()) { + + BlobId id = BlobId.of("b", "o"); + ApiFuture futureBlobDescriptor = storage.blobReadSession(id); + + try (BlobReadSession bd = futureBlobDescriptor.get(5, TimeUnit.SECONDS)) { + // Read 1 (should trigger Redirect 1, then succeed) + byte[] actual1 = + bd.readAs(ReadProjectionConfigs.asFutureBytes().withRangeSpec(RangeSpec.of(10L, 5L))) + .get(1, TimeUnit.SECONDS); + assertThat(xxd(actual1)).isEqualTo(xxd(content1.getBytes())); + + // Read 2 (should trigger Redirect 2, then succeed) + byte[] actual2 = + bd.readAs(ReadProjectionConfigs.asFutureBytes().withRangeSpec(RangeSpec.of(15L, 5L))) + .get(1, TimeUnit.SECONDS); + assertThat(xxd(actual2)).isEqualTo(xxd(content2.getBytes())); + + // Read 3 (should trigger Redirect 3, Redirect 4, Redirect 5, then succeed) + byte[] actual3 = + bd.readAs(ReadProjectionConfigs.asFutureBytes().withRangeSpec(RangeSpec.of(20L, 5L))) + .get(1, TimeUnit.SECONDS); + assertThat(xxd(actual3)).isEqualTo(xxd(content3.getBytes())); + } + } + } + static final class FakeStorage extends StorageImplBase { private final Map>> db; diff --git a/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITRcuBidiReadTest.java b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITRcuBidiReadTest.java new file mode 100644 index 000000000000..1f285e763583 --- /dev/null +++ b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITRcuBidiReadTest.java @@ -0,0 +1,267 @@ +/* + * Copyright 2026 Google LLC + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.google.cloud.storage; + +import static com.google.cloud.storage.TestUtils.xxd; +import static com.google.common.truth.Truth.assertThat; +import static org.junit.Assert.assertThrows; + +import com.google.api.core.ApiFuture; +import com.google.api.gax.rpc.OutOfRangeException; +import com.google.cloud.storage.BlobAppendableUpload.AppendableUploadWriteableByteChannel; +import com.google.cloud.storage.Storage.BlobWriteOption; +import com.google.cloud.storage.TransportCompatibility.Transport; +import com.google.cloud.storage.ZeroCopySupport.DisposableByteString; +import com.google.cloud.storage.it.runner.StorageITRunner; +import com.google.cloud.storage.it.runner.annotations.Backend; +import com.google.cloud.storage.it.runner.annotations.CrossRun; +import com.google.cloud.storage.it.runner.annotations.Inject; +import com.google.cloud.storage.it.runner.annotations.LocationType; +import java.nio.ByteBuffer; +import java.util.Arrays; +import java.util.UUID; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import org.junit.AfterClass; +import org.junit.Before; +import org.junit.Ignore; +import org.junit.Test; +import org.junit.runner.RunWith; + +@RunWith(StorageITRunner.class) +@CrossRun( + backends = {Backend.PROD}, + transports = {Transport.GRPC}, + locations = {LocationType.REGIONAL_RAPID}) +@Ignore("TODO: Enable tests once RCU is available in prod") +public final class ITRcuBidiReadTest { + + @Inject public Storage storage; + @Inject public BucketInfo bucket; + + private static boolean initialized = false; + private static Storage staticStorage; + private static BlobId closeTestBlobId; + private static BlobId zeroCopyTestBlobId; + private static BlobId multipleRangeTestBlobId; + private static BlobId outOfRangeTestBlobId; + + private static byte[] closeTestData; + private static byte[] zeroCopyTestData; + private static byte[] multipleRangeTestData; + private static byte[] outOfRangeTestData; + + @Before + public void setUp() throws Exception { + if (initialized) { + return; + } + staticStorage = storage; + + // Generate test data + closeTestData = DataGenerator.base64Characters().genBytes(5 * 1024 * 1024); + zeroCopyTestData = DataGenerator.base64Characters().genBytes(1024 * 1024); + multipleRangeTestData = DataGenerator.base64Characters().genBytes(2 * 1024 * 1024); + outOfRangeTestData = DataGenerator.base64Characters().genBytes(100 * 1024); + + closeTestBlobId = BlobId.of(bucket.getName(), "test-bidi-read-close-" + UUID.randomUUID()); + zeroCopyTestBlobId = BlobId.of(bucket.getName(), "test-bidi-zero-copy-" + UUID.randomUUID()); + multipleRangeTestBlobId = + BlobId.of(bucket.getName(), "test-bidi-multiple-range-" + UUID.randomUUID()); + outOfRangeTestBlobId = + BlobId.of(bucket.getName(), "test-bidi-out-of-range-" + UUID.randomUUID()); + + createObjectForWarming(closeTestBlobId, closeTestData); + createObjectForWarming(zeroCopyTestBlobId, zeroCopyTestData); + createObjectForWarming(multipleRangeTestBlobId, multipleRangeTestData); + createObjectForWarming(outOfRangeTestBlobId, outOfRangeTestData); + initialized = true; + } + + private void createObjectForWarming(BlobId blobId, byte[] data) throws Exception { + BlobInfo info = + BlobInfo.newBuilder(blobId).setStorageClass(StorageClass.valueOf("RAPID")).build(); + BlobAppendableUploadConfig config = BlobAppendableUploadConfig.of(); + BlobAppendableUpload upload = + storage.blobAppendableUpload(info, config, BlobWriteOption.doesNotExist()); + try (AppendableUploadWriteableByteChannel channel = upload.open()) { + ByteBuffer buffer = ByteBuffer.wrap(data); + while (buffer.hasRemaining()) { + channel.write(buffer); + } + channel.finalizeAndClose(); + } + upload.getResult().get(10, TimeUnit.SECONDS); + } + + @AfterClass + public static void tearDownClass() { + if (staticStorage != null) { + for (BlobId blobId : + Arrays.asList( + closeTestBlobId, zeroCopyTestBlobId, multipleRangeTestBlobId, outOfRangeTestBlobId)) { + if (blobId != null) { + try { + staticStorage.delete(blobId); + } catch (Exception ignored) { + // Ignored: Best effort cleanup + } + } + } + } + } + + @Test + public void readPostStreamClose() throws Exception { + ApiFuture futureSession = storage.blobReadSession(closeTestBlobId); + try (BlobReadSession session = futureSession.get(10, TimeUnit.SECONDS)) { + // Start a future read for the entire object + ApiFuture readFuture = session.readAs(ReadProjectionConfigs.asFutureBytes()); + + // Close the session immediately while the transfer is in flight + session.close(); + + // Resolving the future should now fail since the session is closed + ExecutionException ee = + assertThrows(ExecutionException.class, () -> readFuture.get(5, TimeUnit.SECONDS)); + assertThat(ee).hasCauseThat().isInstanceOf(StorageException.class); + assertThat(ee).hasCauseThat().hasCauseThat().isInstanceOf(AsyncSessionClosedException.class); + } + } + + @Test + public void zeroCopyRangeReads() throws Exception { + // Define 3 non-overlapping ranges + RangeSpec r1 = RangeSpec.of(0, 1000); + RangeSpec r2 = RangeSpec.of(50000, 250000); + RangeSpec r3 = RangeSpec.of(800000, 10000); + + ApiFuture futureSession = storage.blobReadSession(zeroCopyTestBlobId); + try (BlobReadSession session = futureSession.get(10, TimeUnit.SECONDS)) { + // Start concurrent zero-copy range reads + ApiFuture f1 = + session.readAs(ReadProjectionConfigs.asFutureByteString().withRangeSpec(r1)); + ApiFuture f2 = + session.readAs(ReadProjectionConfigs.asFutureByteString().withRangeSpec(r2)); + ApiFuture f3 = + session.readAs(ReadProjectionConfigs.asFutureByteString().withRangeSpec(r3)); + + // Resolve and verify Range 1 + try (DisposableByteString d1 = f1.get(10, TimeUnit.SECONDS)) { + assertThat(d1).isNotNull(); + byte[] expected = Arrays.copyOfRange(zeroCopyTestData, 0, 1000); + assertThat(xxd(d1.byteString().toByteArray())).isEqualTo(xxd(expected)); + } + + // Resolve and verify Range 2 + try (DisposableByteString d2 = f2.get(10, TimeUnit.SECONDS)) { + assertThat(d2).isNotNull(); + byte[] expected = Arrays.copyOfRange(zeroCopyTestData, 50000, 50000 + 250000); + assertThat(xxd(d2.byteString().toByteArray())).isEqualTo(xxd(expected)); + } + + // Resolve and verify Range 3 + try (DisposableByteString d3 = f3.get(10, TimeUnit.SECONDS)) { + assertThat(d3).isNotNull(); + byte[] expected = Arrays.copyOfRange(zeroCopyTestData, 800000, 800000 + 10000); + assertThat(xxd(d3.byteString().toByteArray())).isEqualTo(xxd(expected)); + } + } + } + + @Test + public void multipleRangedRead() throws Exception { + // Define 4 non-overlapping ranges (each 512KB) + int rangeSize = 512 * 1024; + RangeSpec r1 = RangeSpec.of(0, rangeSize); + RangeSpec r2 = RangeSpec.of(rangeSize, rangeSize); + RangeSpec r3 = RangeSpec.of(2 * rangeSize, rangeSize); + RangeSpec r4 = RangeSpec.of(3 * rangeSize, rangeSize); + + ApiFuture futureSession = storage.blobReadSession(multipleRangeTestBlobId); + try (BlobReadSession session = futureSession.get(10, TimeUnit.SECONDS)) { + // Start concurrent byte range reads + ApiFuture f1 = + session.readAs(ReadProjectionConfigs.asFutureBytes().withRangeSpec(r1)); + ApiFuture f2 = + session.readAs(ReadProjectionConfigs.asFutureBytes().withRangeSpec(r2)); + ApiFuture f3 = + session.readAs(ReadProjectionConfigs.asFutureBytes().withRangeSpec(r3)); + ApiFuture f4 = + session.readAs(ReadProjectionConfigs.asFutureBytes().withRangeSpec(r4)); + + // Resolve and verify Range 1 + byte[] b1 = f1.get(10, TimeUnit.SECONDS); + assertThat(xxd(b1)).isEqualTo(xxd(Arrays.copyOfRange(multipleRangeTestData, 0, rangeSize))); + + // Resolve and verify Range 2 + byte[] b2 = f2.get(10, TimeUnit.SECONDS); + assertThat(xxd(b2)) + .isEqualTo(xxd(Arrays.copyOfRange(multipleRangeTestData, rangeSize, 2 * rangeSize))); + + // Resolve and verify Range 3 + byte[] b3 = f3.get(10, TimeUnit.SECONDS); + assertThat(xxd(b3)) + .isEqualTo(xxd(Arrays.copyOfRange(multipleRangeTestData, 2 * rangeSize, 3 * rangeSize))); + + // Resolve and verify Range 4 + byte[] b4 = f4.get(10, TimeUnit.SECONDS); + assertThat(xxd(b4)) + .isEqualTo(xxd(Arrays.copyOfRange(multipleRangeTestData, 3 * rangeSize, 4 * rangeSize))); + } + } + + @Test + public void readFromBucketThatDoesNotExistShouldRaiseStorageException() throws Exception { + String nonExistentBucketName = "java-storage-non-existent-bucket-" + UUID.randomUUID(); + BlobId blobId = BlobId.of(nonExistentBucketName, "someobject"); + + ApiFuture futureObjectReadSession = storage.blobReadSession(blobId); + + ExecutionException ee = + assertThrows( + ExecutionException.class, () -> futureObjectReadSession.get(10, TimeUnit.SECONDS)); + assertThat(ee).hasCauseThat().isInstanceOf(StorageException.class); + StorageException se = (StorageException) ee.getCause(); + assertThat(se.getCode()).isAnyOf(404, 403); + } + + @Test + public void outOfRange() throws Exception { + ApiFuture futureSession = storage.blobReadSession(outOfRangeTestBlobId); + try (BlobReadSession session = futureSession.get(10, TimeUnit.SECONDS)) { + // Start a valid range read on the session first to verify it succeeds + ApiFuture fValid = + session.readAs( + ReadProjectionConfigs.asFutureBytes().withRangeSpec(RangeSpec.of(0, 1000))); + byte[] bytes = fValid.get(10, TimeUnit.SECONDS); + assertThat(xxd(bytes)).isEqualTo(xxd(Arrays.copyOfRange(outOfRangeTestData, 0, 1000))); + + // Start an out-of-bounds range read (offset > size) + ApiFuture fOutOfRange = + session.readAs( + ReadProjectionConfigs.asFutureBytes() + .withRangeSpec(RangeSpec.beginAt(100 * 1024 + 1000))); + + // Verify that resolving it throws OutOfRangeException + ExecutionException ee = + assertThrows(ExecutionException.class, () -> fOutOfRange.get(10, TimeUnit.SECONDS)); + assertThat(ee).hasCauseThat().isInstanceOf(StorageException.class); + assertThat(ee).hasCauseThat().hasCauseThat().isInstanceOf(OutOfRangeException.class); + } + } +}