test(storage): add fast open integration tests

java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITFastOpenTest.java

@@ -0,0 +1,233 @@
+/*
+ * 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 java.util.concurrent.TimeUnit.SECONDS;
+import static java.util.stream.Collectors.toList;
+import static org.junit.Assert.assertThrows;
+
+import com.google.api.gax.rpc.OutOfRangeException;
+import com.google.cloud.storage.BlobAppendableUpload.AppendableUploadWriteableByteChannel;
+import com.google.cloud.storage.Storage.BlobSourceOption;
+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.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.registry.Generator;
+import com.google.cloud.storage.otel.TestExporter;
+import com.google.common.hash.Hashing;
+import io.opentelemetry.api.common.AttributeKey;
+import io.opentelemetry.sdk.OpenTelemetrySdk;
+import io.opentelemetry.sdk.trace.SdkTracerProvider;
+import io.opentelemetry.sdk.trace.export.SimpleSpanProcessor;
+import java.io.ByteArrayOutputStream;
+import java.nio.ByteBuffer;
+import java.nio.channels.Channels;
+import java.nio.channels.ScatteringByteChannel;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.ExecutionException;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+@RunWith(StorageITRunner.class)
+@CrossRun(
+    backends = {Backend.PROD},
+    transports = Transport.GRPC)
+public final class ITFastOpenTest {
+
+  private static final int OBJECT_SIZE = 512 * 1024;
+  private static final ReadAsFutureBytes READ_10_TO_20 =
+      ReadProjectionConfigs.asFutureBytes().withRangeSpec(RangeSpec.of(10, 10));
+
+  @Inject public Storage storage;
+
+  @Inject
+  @BucketFixture(BucketType.RAPID)
+  public BucketInfo bucket;
+
+  @Inject public Generator generator;
+
+  private ChecksummedTestContent content;
+  private BlobId blobId;
+
+  @Before
+  public void setUp() throws Exception {
+    content = ChecksummedTestContent.of(DataGenerator.base64Characters().genBytes(OBJECT_SIZE));
+    blobId = create(content);
+  }
+
+  @Test
+  public void bytes_areServedFromTheOpen() throws Exception {
+    TestExporter exporter = new TestExporter();
+    OpenTelemetrySdk otel =
+        OpenTelemetrySdk.builder()
+            .setTracerProvider(
+                SdkTracerProvider.builder()
+                    .addSpanProcessor(SimpleSpanProcessor.create(exporter))
+                    .build())
+            .build();
+
+    try (Storage traced =
+            storage.getOptions().toBuilder().setOpenTelemetry(otel).build().getService();
+        BlobReadSession session = traced.blobReadSession(blobId, READ_10_TO_20).get(30, SECONDS)) {
+      byte[] actual = session.readAs(READ_10_TO_20).get(30, SECONDS);
+      assertThat(xxd(actual)).isEqualTo(xxd(content.getBytes(10, 10)));
+    }
+
+    List<String> cacheStatus =
+        exporter.getExportedSpans().stream()
+            .map(s -> s.getAttributes().get(AttributeKey.stringKey("gcs.bidi.cache_status")))
+            .filter(Objects::nonNull)
+            .collect(toList());
+    assertThat(cacheStatus).containsExactly("HIT");
+  }
+
+  @Test
+  public void byteString() throws Exception {
+    ReadAsFutureByteString config =
+        ReadProjectionConfigs.asFutureByteString().withRangeSpec(RangeSpec.of(10, 10));
+
+    try (BlobReadSession session = storage.blobReadSession(blobId, config).get(30, SECONDS);
+        DisposableByteString actual = session.readAs(config).get(30, SECONDS)) {
+      assertThat(xxd(actual.byteString().toByteArray())).isEqualTo(xxd(content.getBytes(10, 10)));
+    }
+  }
+
+  @Test
+  public void channel_wholeObject() throws Exception {
+    ReadAsChannel config = ReadProjectionConfigs.asChannel();
+
+    ByteArrayOutputStream actual = new ByteArrayOutputStream();
+    try (BlobReadSession session = storage.blobReadSession(blobId, config).get(30, SECONDS);
+        ScatteringByteChannel channel = session.readAs(config)) {
+      Buffers.copyUsingBuffer(ByteBuffer.allocate(64 * 1024), channel, Channels.newChannel(actual));
+    }
+
+    assertThat(xxd(actual.toByteArray())).isEqualTo(xxd(content.getBytes()));
+  }
+
+  @Test
+  public void moreReadsOnTheSameSession() throws Exception {
+    ReadAsFutureBytes read20To30 =
+        ReadProjectionConfigs.asFutureBytes().withRangeSpec(RangeSpec.of(20, 10));
+
+    try (BlobReadSession session =
+        storage.blobReadSession(blobId, READ_10_TO_20).get(30, SECONDS)) {
+      byte[] fastOpen = session.readAs(READ_10_TO_20).get(30, SECONDS);
+      byte[] other = session.readAs(read20To30).get(30, SECONDS);
+      byte[] again = session.readAs(READ_10_TO_20).get(30, SECONDS);
+      assertThat(xxd(fastOpen)).isEqualTo(xxd(content.getBytes(10, 10)));
+      assertThat(xxd(other)).isEqualTo(xxd(content.getBytes(20, 10)));
+      assertThat(xxd(again)).isEqualTo(xxd(content.getBytes(10, 10)));
+    }
+  }
+
+  @Test
+  public void outOfRange_theOpenSucceeds_andReadAsFails() throws Exception {
+    ReadAsFutureBytes config =
+        ReadProjectionConfigs.asFutureBytes().withRangeSpec(RangeSpec.beginAt(OBJECT_SIZE + 1));
+
+    try (BlobReadSession session = storage.blobReadSession(blobId, config).get(30, SECONDS)) {
+      ExecutionException e =
+          assertThrows(ExecutionException.class, () -> session.readAs(config).get(30, SECONDS));
+      assertThat(e).hasCauseThat().hasCauseThat().isInstanceOf(OutOfRangeException.class);
+    }
+  }
+
+  @Test
+  public void largeRead_spanningManyResponses() throws Exception {
+    ChecksummedTestContent large =
+        ChecksummedTestContent.of(DataGenerator.base64Characters().genBytes(16 * 1024 * 1024));
+    BlobId largeBlobId = create(large);
+    ReadAsFutureBytes config = ReadProjectionConfigs.asFutureBytes();
+
+    try (BlobReadSession session = storage.blobReadSession(largeBlobId, config).get(30, SECONDS)) {
+      byte[] actual = session.readAs(config).get(30, SECONDS);
+      assertThat(actual.length).isEqualTo(large.getBytes().length);
+      assertThat(Hashing.crc32c().hashBytes(actual).asInt()).isEqualTo(large.getCrc32c());
+    }
+  }
+
+  @Test
+  public void readStartingAtTheEnd_isEmpty() throws Exception {
+    ReadAsFutureBytes config =
+        ReadProjectionConfigs.asFutureBytes().withRangeSpec(RangeSpec.beginAt(OBJECT_SIZE));
+
+    try (BlobReadSession session = storage.blobReadSession(blobId, config).get(30, SECONDS)) {
+      assertThat(session.readAs(config).get(30, SECONDS)).isEmpty();
+    }
+  }
+
+  @Test
+  public void objectDoesNotExist_theOpenFailsWith404() {
+    BlobId missing = BlobId.of(bucket.getName(), generator.randomObjectName());
+
+    ExecutionException e =
+        assertThrows(
+            ExecutionException.class,
+            () -> storage.blobReadSession(missing, READ_10_TO_20).get(30, SECONDS));
+    assertThat(e).hasCauseThat().isInstanceOf(StorageException.class);
+    assertThat(((StorageException) e.getCause()).getCode()).isEqualTo(404);
+  }
+
+  @Test
+  public void preconditionFails_theOpenFailsWith412() {
+    BlobId withoutGeneration = BlobId.of(blobId.getBucket(), blobId.getName());
+    BlobSourceOption wrongGeneration = BlobSourceOption.generationMatch(blobId.getGeneration() + 1);
+
+    ExecutionException e =
+        assertThrows(
+            ExecutionException.class,
+            () ->
+                storage
+                    .blobReadSession(withoutGeneration, READ_10_TO_20, wrongGeneration)
+                    .get(30, SECONDS));
+    assertThat(e).hasCauseThat().isInstanceOf(StorageException.class);
+    assertThat(((StorageException) e.getCause()).getCode()).isEqualTo(412);
+  }
+
+  @Test
+  public void close_withoutClaimingTheFastOpenRead() throws Exception {
+    BlobReadSession session = storage.blobReadSession(blobId, READ_10_TO_20).get(30, SECONDS);
+
+    session.close();
+
+    assertThrows(IllegalStateException.class, () -> session.readAs(READ_10_TO_20));
+  }
+
+  private BlobId create(ChecksummedTestContent content) throws Exception {
+    BlobInfo info = BlobInfo.newBuilder(bucket, generator.randomObjectName()).build();
+    BlobAppendableUpload upload =
+        storage.blobAppendableUpload(
+            info, BlobAppendableUploadConfig.of(), BlobWriteOption.doesNotExist());
+    try (AppendableUploadWriteableByteChannel channel = upload.open()) {
+      Buffers.emptyTo(ByteBuffer.wrap(content.getBytes()), channel);
+      channel.finalizeAndClose();
+    }
+    return upload.getResult().get(10, SECONDS).getBlobId();
+  }
+}