Skip to content

test(storage): Add RCU Bidi Read and Bidi Write integration tests #14531

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
nidhiii-27 wants to merge 6 commits into from
+446 −33
Open

test(storage): Add RCU Bidi Read and Bidi Write integration tests #14531

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
8b1530c
test(storage): Add RCU Bidi Read and Bidi Write integration tests
nidhiii-27 Sep 28, 2026
2ff3a7f
style(storage): format ITObjectReadSessionFakeTest with fmt-maven-plugin
nidhiii-27 Sep 28, 2026
a1af012
test(storage): address PR review comments for RCU bidi read and write…
nidhiii-27 Sep 29, 2026
81f8478
fix(storage): address PR review feedback and consolidate RCU Bidi tests
nidhiii-27 Nov 3, 2026
7ae63aa
test(storage): consolidate RCU Bidi Read ITs into ITObjectReadSession…
nidhiii-27 Nov 7, 2026
7069088
test(storage): reuse withRedirectToken in ITAppendableUploadFakeTest
nidhiii-27 Nov 9, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
170 changes: 170 additions & 0 deletions ...ogle-cloud-storage/src/test/java/com/google/cloud/storage/ITAppendableUploadFakeTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 =

@Dhriti07 Dhriti07 Nov 9, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The five hand-built reconnect_tokenN requests repeat what bidiWriteObjectRedirectedError_maxAttempts already does with BidiUploadTestUtils.withRedirectToken(base, token). Can we reuse it ?

@nidhiii-27 nidhiii-27 Nov 9, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done. Reused BidiUploadTestUtils.withRedirectToken(base, token) for reconnect_token1 through reconnect_token5.

Co-authored by AI Agent

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<StreamObserver<BidiWriteObjectResponse>> defHandler =
respond -> {
if (defCounter.getAndIncrement() == 0) {
respond.onError(packRedirectIntoAbortedException(makeRedirect(routingToken1)));
} else {
respond.onNext(res_def_success);
}
};

AtomicInteger ghiCounter = new AtomicInteger();
Consumer<StreamObserver<BidiWriteObjectResponse>> ghiHandler =
respond -> {
if (ghiCounter.getAndIncrement() == 0) {
respond.onError(packRedirectIntoAbortedException(makeRedirect(routingToken2)));
} else {
respond.onNext(res_ghi_success);
}
};

AtomicInteger jCounter = new AtomicInteger();
Consumer<StreamObserver<BidiWriteObjectResponse>> jHandler =
respond -> {
if (jCounter.getAndIncrement() == 0) {
respond.onError(packRedirectIntoAbortedException(makeRedirect(routingToken3)));
} else {
respond.onNext(res_j_final);
respond.onCompleted();
}
};

FakeStorage fake =
FakeStorage.of(
ImmutableMap
.<BidiWriteObjectRequest, Consumer<StreamObserver<BidiWriteObjectResponse>>>
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<BlobInfo> 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".
Expand Down
87 changes: 62 additions & 25 deletions ...e/google-cloud-storage/src/test/java/com/google/cloud/storage/ITAppendableUploadTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;

Expand All @@ -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();

Expand All @@ -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();
Expand Down Expand Up @@ -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();
Expand Down Expand Up @@ -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());
Expand All @@ -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());
Expand Down Expand Up @@ -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);
Expand All @@ -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());
Expand All @@ -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);
Expand All @@ -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);
Expand All @@ -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);
Expand Down Expand Up @@ -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());
Expand All @@ -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()) {
Expand Down Expand Up @@ -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());
Expand All @@ -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);
Expand All @@ -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());
Expand All @@ -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);
Expand Down
Loading
Loading