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..d94534f966fd 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,176 @@ 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(); + + BidiWriteObjectRequest redirectReconcile1 = + reconnect.toBuilder() + .setAppendObjectSpec( + reconnect.getAppendObjectSpec().toBuilder().setWriteHandle(writeHandle1)) + .build(); + BidiWriteObjectRequest redirectReconcile2 = + reconnect.toBuilder() + .setAppendObjectSpec( + reconnect.getAppendObjectSpec().toBuilder().setWriteHandle(writeHandle2)) + .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 = + BidiUploadTestUtils.withRedirectToken(redirectReconcile1, routingToken1); + 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 = + BidiUploadTestUtils.withRedirectToken(redirectReconcile1, routingToken2); + BidiWriteObjectResponse res_lookup_token2 = + BidiWriteObjectResponse.newBuilder() + .setWriteHandle(writeHandle2) + .setPersistedSize(6) + .build(); + + // 6. Fourth write of "J" (finalize/finish write) + BidiWriteObjectResponse res_j_final = + resource_10.toBuilder().setWriteHandle(writeHandle3).build(); + + // 7. Reconnect to routingToken3 + BidiWriteObjectRequest reconnect_token3 = + BidiUploadTestUtils.withRedirectToken(redirectReconcile2, routingToken3); + + // 8. Reconnect to routingToken4 + BidiWriteObjectRequest reconnect_token4 = + BidiUploadTestUtils.withRedirectToken(redirectReconcile2, routingToken4); + + // 9. Reconnect to routingToken5 + BidiWriteObjectRequest reconnect_token5 = + BidiUploadTestUtils.withRedirectToken(redirectReconcile2, routingToken5); + 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(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 d278a24ee77f..da2c89b5f06e 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; @@ -60,17 +58,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; @@ -87,7 +84,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(); @@ -111,7 +111,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(); @@ -144,7 +147,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(); @@ -182,7 +188,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()); @@ -193,7 +201,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()); @@ -222,7 +233,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); @@ -244,7 +257,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()); @@ -255,7 +270,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); @@ -274,7 +292,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); @@ -298,7 +318,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); @@ -331,7 +353,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()); @@ -342,7 +366,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()) { @@ -372,7 +399,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()); @@ -382,7 +411,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); @@ -409,7 +441,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()); @@ -419,7 +453,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/ITObjectReadSessionTest.java b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITObjectReadSessionTest.java index cd3eb3170d3d..22d5b79bf7f2 100644 --- a/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITObjectReadSessionTest.java +++ b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITObjectReadSessionTest.java @@ -35,10 +35,9 @@ 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.registry.Generator; import com.google.common.base.Stopwatch; import com.google.common.hash.Hasher; @@ -65,16 +64,17 @@ @RunWith(StorageITRunner.class) @CrossRun( backends = {Backend.TEST_BENCH}, - 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}) public final class ITObjectReadSessionTest { private static final int _512KiB = 512 * 1024; @Inject public Storage storage; - @Inject - @BucketFixture(BucketType.RAPID) - public BucketInfo bucket; + @Inject public BucketInfo bucket; @Inject public Generator generator; @@ -356,7 +356,10 @@ public void outOfRange() private BlobInfo create(ChecksummedTestContent content) throws IOException, ExecutionException, InterruptedException, TimeoutException { - BlobInfo info = BlobInfo.newBuilder(bucket, generator.randomObjectName()).build(); + BlobInfo info = + BlobInfo.newBuilder(bucket, generator.randomObjectName()) + .setStorageClass(StorageClass.valueOf("RAPID")) + .build(); BlobAppendableUploadConfig config = BlobAppendableUploadConfig.of(); if (backend == Backend.TEST_BENCH) {