Skip to content

chore(storage): remove the fast open prototype #14874

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

Draft
Dhriti07 wants to merge 1 commit into
base: fast-open/04-failure-retry-tests
Choose a base branch
from
Draft

chore(storage): remove the fast open prototype #14874

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
8 changes: 1 addition & 7 deletions .../google-cloud-storage/src/main/java/com/google/cloud/storage/ObjectReadSessionStream.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,6 @@
import com.google.cloud.storage.ObjectReadSessionState.OpenArguments;
import com.google.cloud.storage.ResponseContentLifecycleHandle.ChildRef;
import com.google.cloud.storage.RetryContext.OnSuccess;
import com.google.cloud.storage.StorageDataClient.Borrowable;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.protobuf.ByteString;
Expand All @@ -58,7 +57,7 @@
import org.checkerframework.checker.nullness.qual.Nullable;

final class ObjectReadSessionStream
implements ClientStream<BidiReadObjectRequest>, ApiFuture<Void>, IOAutoCloseable, Borrowable {
implements ClientStream<BidiReadObjectRequest>, ApiFuture<Void>, IOAutoCloseable {

private final SettableApiFuture<Void> objectReadSessionResolveFuture;

Expand Down Expand Up @@ -206,11 +205,6 @@ boolean isOpen() {
return openLeases.get() > 0;
}

public void borrow() {
checkOpen();
openLeases.incrementAndGet();
}

private void checkOpen() {
Preconditions.checkState(isOpen(), "Stream closed");
}
Expand Down
97 changes: 0 additions & 97 deletions ...torage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageDataClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,14 +23,11 @@
import com.google.api.core.InternalApi;
import com.google.api.gax.grpc.GrpcCallContext;
import com.google.cloud.storage.GrpcUtils.ZeroCopyBidiStreamingCallable;
import com.google.cloud.storage.ObjectReadSessionState.OpenArguments;
import com.google.cloud.storage.ReadProjectionConfig.ProjectionType;
import com.google.cloud.storage.RetryContext.RetryContextProvider;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.storage.v2.BidiReadObjectRequest;
import com.google.storage.v2.BidiReadObjectResponse;
import java.io.IOException;
import java.time.Duration;
import java.util.HashMap;
import java.util.List;
Expand Down Expand Up @@ -117,46 +114,6 @@ ApiFuture<ObjectReadSession> readSession(
return objectReadSessionFuture;
}

<Projection> ApiFuture<FastOpenObjectReadSession<Projection>> fastOpenReadSession(
BidiReadObjectRequest openRequest,
GrpcCallContext ctx,
ReadProjectionConfig<Projection> config) {
checkArgument(
openRequest.getReadRangesList().isEmpty(),
"ranges included in the initial request are not supported");
checkArgument(
config.getType() == ProjectionType.STREAM_READ,
"unsupported ReadProjectionConfig: %s",
config.getClass().getName());
ObjectReadSessionState state = new ObjectReadSessionState(ctx, openRequest);

ZeroCopyBidiStreamingCallable<BidiReadObjectRequest, BidiReadObjectResponse> callable =
getCallable();
ObjectReadSessionStream stream =
ObjectReadSessionStream.create(executor, callable, state, retryContextProvider.create());

long readId = state.newReadId();
ObjectReadSessionStreamRead<Projection> read =
config.cast().newRead(readId, retryContextProvider.create());
state.putOutstandingRead(readId, read);

ApiFuture<FastOpenObjectReadSession<Projection>> objectReadSessionFuture =
ApiFutures.transform(
stream,
nowOpen ->
new FastOpenObjectReadSession<>(
new ObjectReadSessionImpl(
executor, callable, stream, state, retryContextProvider, ImmutableMap.of()),
read,
stream),
executor);
OpenArguments openArguments = state.getOpenArguments();
BidiReadObjectRequest req = openArguments.getReq();
stream.send(req);
read.setOnCloseCallback(stream);
return objectReadSessionFuture;
}

@SuppressWarnings("ResultOfMethodCallIgnored")
@Override
public void close() throws Exception {
Expand All @@ -182,58 +139,4 @@ static StorageDataClient create(
return new StorageDataClient(
executor, terminationAwaitDuration, read, retryContextProvider, onClose);
}

@FunctionalInterface
interface Borrowable {
void borrow();
}

static final class FastOpenObjectReadSession<Projection> implements IOAutoCloseable {
private final ObjectReadSession session;
private final ObjectReadSessionStreamRead<Projection> read;
private final Borrowable borrowable;
private boolean sessionLeased;

private FastOpenObjectReadSession(
ObjectReadSession session,
ObjectReadSessionStreamRead<Projection> read,
Borrowable borrowable) {
this.session = session;
this.read = read;
this.borrowable = borrowable;
this.sessionLeased = false;
}

ObjectReadSession getSession() {
if (!sessionLeased) {
sessionLeased = true;
borrowable.borrow();
}
return session;
}

ObjectReadSessionStreamRead<Projection> getRead() {
return read;
}

Projection getProjection() {
return read.project();
}

@Override
public void close() throws IOException {
//noinspection EmptyTryBlock
try (IOAutoCloseable ignore1 = session;
IOAutoCloseable ignore2 = read) {
// use try-with to ensure full cleanup
}
}

public static <Projection> FastOpenObjectReadSession<Projection> of(
ObjectReadSession session,
ObjectReadSessionStreamRead<Projection> read,
Borrowable borrowable) {
return new FastOpenObjectReadSession<>(session, read, borrowable);
}
}
}
19 changes: 8 additions & 11 deletions ...gle-cloud-storage/src/test/java/com/google/cloud/storage/ITObjectReadSessionFakeTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,6 @@
import com.google.cloud.storage.Hasher.UncheckedChecksumMismatchException;
import com.google.cloud.storage.OtelStorageDecorator.OtelDecoratingBlobReadSession;
import com.google.cloud.storage.Storage.BlobSourceOption;
import com.google.cloud.storage.StorageDataClient.FastOpenObjectReadSession;
import com.google.cloud.storage.ZeroCopySupport.DisposableByteString;
import com.google.cloud.storage.it.ChecksummedTestContent;
import com.google.cloud.storage.it.GrpcPlainRequestLoggingInterceptor;
Expand Down Expand Up @@ -1310,7 +1309,7 @@ public void onCompleted() {
}

@Test
public void gettingSessionFromFastOpenKeepsTheSessionOpenUntilClosed() throws Exception {
public void fastOpenRead_thenMoreReadsOnTheSameSession() throws Exception {
ChecksummedTestContent expected = ChecksummedTestContent.of(ALL_OBJECT_BYTES, 10, 30);

ChecksummedTestContent content1 = ChecksummedTestContent.of(ALL_OBJECT_BYTES, 10, 10);
Expand Down Expand Up @@ -1384,19 +1383,17 @@ public void gettingSessionFromFastOpenKeepsTheSessionOpenUntilClosed() throws Ex
.build())
.build();

ApiFuture<FastOpenObjectReadSession<ApiFuture<DisposableByteString>>> future =
dataClient.fastOpenReadSession(
req,
GrpcCallContext.createDefault(),
ReadProjectionConfigs.asFutureByteString().withRangeSpec(RangeSpec.of(10, 10)));
ReadAsFutureByteString fastOpenConfig =
ReadProjectionConfigs.asFutureByteString().withRangeSpec(RangeSpec.of(10, 10));
Comment on lines +1386 to +1387

@gemini-code-assist gemini-code-assist Bot Oct 31, 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.

medium

To align with the interface-based design of the SDK and prevent potential compilation or type-mismatch issues, it is recommended to declare fastOpenConfig using the ReadProjectionConfig<ApiFuture<DisposableByteString>> interface type rather than the concrete implementation class.

Suggested change
ReadAsFutureByteString fastOpenConfig =
ReadProjectionConfigs.asFutureByteString().withRangeSpec(RangeSpec.of(10, 10));
ReadProjectionConfig<ApiFuture<DisposableByteString>> fastOpenConfig =
ReadProjectionConfigs.asFutureByteString().withRangeSpec(RangeSpec.of(10, 10));

ApiFuture<ObjectReadSession> future =
dataClient.readSession(
req, GrpcCallContext.createDefault(), ImmutableList.of(fastOpenConfig));

ByteString bytes = ByteString.empty();
Exception caught = null;

try (FastOpenObjectReadSession<ApiFuture<DisposableByteString>> fastOpenChannel =
future.get(5, TimeUnit.SECONDS);
ObjectReadSession session = fastOpenChannel.getSession()) {
ApiFuture<DisposableByteString> futureBytes1 = fastOpenChannel.getProjection();
try (ObjectReadSession session = future.get(5, TimeUnit.SECONDS)) {
ApiFuture<DisposableByteString> futureBytes1 = session.readAs(fastOpenConfig);
try (DisposableByteString disposableByteString = futureBytes1.get()) {
bytes = bytes.concat(disposableByteString.byteString());
}
Expand Down
10 changes: 5 additions & 5 deletions ...gle-cloud-storage/src/test/java/com/google/cloud/storage/ITStorageDataClientFakeTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,9 +24,9 @@
import com.google.api.core.ApiFuture;
import com.google.api.gax.grpc.GrpcCallContext;
import com.google.cloud.storage.ITObjectReadSessionFakeTest.FakeStorage;
import com.google.cloud.storage.StorageDataClient.FastOpenObjectReadSession;
import com.google.cloud.storage.ZeroCopySupport.DisposableByteString;
import com.google.cloud.storage.it.ChecksummedTestContent;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.io.ByteStreams;
import com.google.protobuf.ByteString;
Expand Down Expand Up @@ -137,10 +137,10 @@ private <P> void doTest(ReadProjectionConfig<P> config, ThrowableFunction<P, byt
.build())
.build();

ApiFuture<FastOpenObjectReadSession<P>> f =
dataClient.fastOpenReadSession(req, GrpcCallContext.createDefault(), config);
try (FastOpenObjectReadSession<P> fastOpen = f.get(3, TimeUnit.SECONDS)) {
byte[] apply = func.apply(fastOpen.getProjection());
ApiFuture<ObjectReadSession> f =
dataClient.readSession(req, GrpcCallContext.createDefault(), ImmutableList.of(config));
try (ObjectReadSession session = f.get(3, TimeUnit.SECONDS)) {
byte[] apply = func.apply(session.readAs(config));
assertThat(xxd(apply)).isEqualTo(xxd(content.getBytes()));
}
}
Expand Down
Loading