From 0361672a9ba30520d7c5c9607da05e142b7d2ef7 Mon Sep 17 00:00:00 2001 From: Shunping Huang Date: Wed, 16 Sep 2026 21:35:51 -0400 Subject: [PATCH 1/9] Route the remaining GCS calls through GcsUtilV2 The facade only had V2 branches on its GcsPath-typed methods, but GcsFileSystem and GcpOptions call the String-typed and legacy-model ones, so match, list, copy, rename, delete and bucket creation all stayed on V1 even with use_gcsutil_v2 set. Add V2 branches to those, converting Blob and BucketInfo back to the JSON API model so their callers need no change. Strategies are picked to match V1: copy and rename rewrite without a destination precondition, remove ignores a 404, and buckets are created with projectPrivate ACLs. --- .../beam/sdk/extensions/gcp/util/GcsUtil.java | 178 +++++++++++++++- .../sdk/extensions/gcp/util/GcsUtilV2.java | 23 +- .../sdk/extensions/gcp/util/GcsUtilTest.java | 200 ++++++++++++++++++ 3 files changed, 396 insertions(+), 5 deletions(-) diff --git a/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtil.java b/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtil.java index 9e2d82ae1eb4..908981d3bc87 100644 --- a/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtil.java +++ b/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtil.java @@ -17,6 +17,7 @@ */ package org.apache.beam.sdk.extensions.gcp.util; +import com.google.api.client.util.DateTime; import com.google.api.gax.paging.Page; import com.google.api.services.storage.model.Bucket; import com.google.api.services.storage.model.Objects; @@ -28,10 +29,17 @@ import com.google.cloud.storage.Storage.BlobSourceOption; import com.google.cloud.storage.Storage.BlobWriteOption; import com.google.cloud.storage.Storage.BucketGetOption; +import com.google.cloud.storage.Storage.BucketTargetOption; +import com.google.cloud.storage.Storage.PredefinedAcl; +import com.google.cloud.storage.StorageClass; import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; import java.io.IOException; +import java.math.BigInteger; import java.nio.channels.SeekableByteChannel; import java.nio.channels.WritableByteChannel; +import java.time.Duration; +import java.time.OffsetDateTime; +import java.util.ArrayList; import java.util.Collection; import java.util.List; import java.util.Set; @@ -151,6 +159,9 @@ public long fileSize(GcsPath path) throws IOException { */ @Deprecated public StorageObject getObject(GcsPath gcsPath) throws IOException { + if (delegateV2 != null) { + return toStorageObject(delegateV2.getBlob(gcsPath)); + } return delegate.getObject(gcsPath); } @@ -166,6 +177,21 @@ public Blob getBlob(GcsPath gcsPath, BlobGetOption... options) throws IOExceptio */ @Deprecated public List getObjects(List gcsPaths) throws IOException { + if (delegateV2 != null) { + List results = new ArrayList<>(); + for (BlobResult blobResult : delegateV2.getBlobs(gcsPaths)) { + Blob blob = blobResult.blob(); + IOException ioException = blobResult.ioException(); + if (blob != null) { + results.add(StorageObjectOrIOException.create(toStorageObject(blob))); + } else if (ioException != null) { + results.add(StorageObjectOrIOException.create(ioException)); + } else { + throw new IOException("Invalid blob result: it holds neither a blob nor an error."); + } + } + return results; + } List legacy = delegate.getObjects(gcsPaths); return legacy.stream() .map(StorageObjectOrIOException::fromLegacy) @@ -186,6 +212,9 @@ public List getBlobs(Iterable gcsPaths, BlobGetOption... op @Deprecated public Objects listObjects(String bucket, String prefix, @Nullable String pageToken) throws IOException { + if (delegateV2 != null) { + return toObjects(delegateV2.listBlobs(bucket, prefix, pageToken)); + } return delegate.listObjects(bucket, prefix, pageToken); } @@ -196,6 +225,9 @@ public Objects listObjects(String bucket, String prefix, @Nullable String pageTo public Objects listObjects( String bucket, String prefix, @Nullable String pageToken, @Nullable String delimiter) throws IOException { + if (delegateV2 != null) { + return toObjects(delegateV2.listBlobs(bucket, prefix, pageToken, delimiter)); + } return delegate.listObjects(bucket, prefix, pageToken, delimiter); } @@ -341,16 +373,27 @@ public long bucketOwner(GcsPath path) throws IOException { } /** - * @deprecated use {@link #createBucket(BucketInfo)}. + * @deprecated use {@link #createBucket(BucketInfo, BucketTargetOption...)}. */ @Deprecated public void createBucket(String projectId, Bucket bucket) throws IOException { + if (delegateV2 != null) { + // GcsUtilV1 always creates buckets with projectPrivate ACLs, which java-storage does not do + // on its own, so they have to be requested explicitly to keep the same access. + delegateV2.createBucket( + projectId, + toBucketInfo(bucket), + BucketTargetOption.predefinedAcl(PredefinedAcl.PROJECT_PRIVATE), + BucketTargetOption.predefinedDefaultObjectAcl(PredefinedAcl.PROJECT_PRIVATE)); + return; + } delegate.createBucket(projectId, bucket); } - public void createBucket(BucketInfo bucketInfo) throws IOException { + public void createBucket(BucketInfo bucketInfo, BucketTargetOption... options) + throws IOException { if (delegateV2 != null) { - delegateV2.createBucket(bucketInfo); + delegateV2.createBucket(bucketInfo, options); } else { throw new IOException("GcsUtil V2 not initialized."); } @@ -377,6 +420,10 @@ public void createBucket(BucketInfo bucketInfo) throws IOException { */ @Deprecated public void removeBucket(Bucket bucket) throws IOException { + if (delegateV2 != null) { + delegateV2.removeBucket(toBucketInfo(bucket)); + return; + } delegate.removeBucket(bucket); } @@ -390,6 +437,14 @@ public void removeBucket(BucketInfo bucketInfo) throws IOException { public void copy(Iterable srcFilenames, Iterable destFilenames) throws IOException { + if (delegateV2 != null) { + // GcsUtilV1 issues a rewrite without any destination precondition, so ALWAYS_OVERWRITE is + // the strategy that preserves its behavior. The strategies that inspect the destination + // would also cost an extra GET per file. + delegateV2.copy( + toGcsPaths(srcFilenames), toGcsPaths(destFilenames), OverwriteStrategy.ALWAYS_OVERWRITE); + return; + } delegate.copy(srcFilenames, destFilenames); } @@ -412,6 +467,22 @@ public void copy( public void rename( Iterable srcFilenames, Iterable destFilenames, MoveOptions... moveOptions) throws IOException { + GcsUtilV2 v2 = delegateV2; + if (v2 != null) { + Set moveOptionSet = Sets.newHashSet(moveOptions); + // Note this differs from renameV2, which defaults to SAFE_OVERWRITE. GcsUtilV1 rewrites + // without a destination precondition, so ALWAYS_OVERWRITE is the behavior preserving choice. + v2.move( + toGcsPaths(srcFilenames), + toGcsPaths(destFilenames), + moveOptionSet.contains(StandardMoveOptions.IGNORE_MISSING_FILES) + ? MissingStrategy.SKIP_IF_MISSING + : MissingStrategy.FAIL_IF_MISSING, + moveOptionSet.contains(StandardMoveOptions.SKIP_IF_DESTINATION_EXISTS) + ? OverwriteStrategy.SKIP_IF_EXISTS + : OverwriteStrategy.ALWAYS_OVERWRITE); + return; + } delegate.rename(srcFilenames, destFilenames, moveOptions); } @@ -453,6 +524,11 @@ public void rename( } public void remove(Collection filenames) throws IOException { + if (delegateV2 != null) { + // GcsUtilV1 ignores a 404 on delete, which is SKIP_IF_MISSING. + delegateV2.remove(toGcsPaths(filenames), MissingStrategy.SKIP_IF_MISSING); + return; + } delegate.remove(filenames); } @@ -470,6 +546,102 @@ public void remove(Iterable paths, MissingStrategy strategy) throws IOE } } + private static List toGcsPaths(Iterable filenames) { + List paths = new ArrayList<>(); + for (String filename : filenames) { + paths.add(GcsPath.fromUri(filename)); + } + return paths; + } + + /** + * Converts a JSON API {@link Bucket} into the java-storage {@link BucketInfo} model. + * + *

Only the properties that callers of the deprecated {@link #createBucket(String, Bucket)} set + * are carried over. Like {@link #toStorageObject}, this is expected to go away with the + * deprecated methods it serves. + */ + private static BucketInfo toBucketInfo(Bucket bucket) { + BucketInfo.Builder builder = BucketInfo.newBuilder(bucket.getName()); + if (bucket.getLocation() != null) { + builder.setLocation(bucket.getLocation()); + } + if (bucket.getStorageClass() != null) { + builder.setStorageClass(StorageClass.valueOf(bucket.getStorageClass())); + } + Bucket.SoftDeletePolicy softDeletePolicy = bucket.getSoftDeletePolicy(); + if (softDeletePolicy != null && softDeletePolicy.getRetentionDurationSeconds() != null) { + builder.setSoftDeletePolicy( + BucketInfo.SoftDeletePolicy.newBuilder() + .setRetentionDuration( + Duration.ofSeconds(softDeletePolicy.getRetentionDurationSeconds())) + .build()); + } + return builder.build(); + } + + /** + * Converts a java-storage {@link Blob} back into the JSON API {@link StorageObject} model. + * + *

This lets the deprecated, legacy typed methods of this class be served by {@link GcsUtilV2} + * without their callers having to change. It is expected to go away once those methods do. + */ + private static StorageObject toStorageObject(Blob blob) { + StorageObject storageObject = + new StorageObject() + .setBucket(blob.getBucket()) + .setName(blob.getName()) + .setGeneration(blob.getGeneration()) + .setMetageneration(blob.getMetageneration()) + .setContentType(blob.getContentType()) + .setContentEncoding(blob.getContentEncoding()) + .setMd5Hash(blob.getMd5()) + .setCrc32c(blob.getCrc32c()) + .setEtag(blob.getEtag()); + Long size = blob.getSize(); + if (size != null) { + storageObject.setSize(BigInteger.valueOf(size)); + } + OffsetDateTime updated = blob.getUpdateTimeOffsetDateTime(); + if (updated != null) { + storageObject.setUpdated(new DateTime(updated.toInstant().toEpochMilli())); + } + OffsetDateTime created = blob.getCreateTimeOffsetDateTime(); + if (created != null) { + storageObject.setTimeCreated(new DateTime(created.toInstant().toEpochMilli())); + } + return storageObject; + } + + /** Converts a single page of java-storage {@link Blob}s into the JSON API {@link Objects}. */ + private static Objects toObjects(Page page) { + List items = new ArrayList<>(); + List prefixes = new ArrayList<>(); + for (Blob blob : page.getValues()) { + // A delimited listing reports each common prefix as a directory placeholder. + if (blob.isDirectory()) { + prefixes.add(blob.getName()); + } else { + items.add(toStorageObject(blob)); + } + } + Objects objects = new Objects(); + // Leave items and prefixes null when empty, as the JSON API does, so that callers looping on + // getItems() != null keep terminating. + if (!items.isEmpty()) { + objects.setItems(items); + } + if (!prefixes.isEmpty()) { + objects.setPrefixes(prefixes); + } + // Page.getNextPageToken() may be an empty string rather than null on the last page. + String nextPageToken = page.hasNextPage() ? page.getNextPageToken() : null; + if (nextPageToken != null) { + objects.setNextPageToken(nextPageToken); + } + return objects; + } + @SuppressFBWarnings("NM_CLASS_NOT_EXCEPTION") public static class StorageObjectOrIOException { final GcsUtilV1.StorageObjectOrIOException delegate; diff --git a/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilV2.java b/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilV2.java index 9d223f810507..8f72ced80f87 100644 --- a/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilV2.java +++ b/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilV2.java @@ -45,6 +45,7 @@ import com.google.cloud.storage.Storage.BlobWriteOption; import com.google.cloud.storage.Storage.BucketField; import com.google.cloud.storage.Storage.BucketGetOption; +import com.google.cloud.storage.Storage.BucketTargetOption; import com.google.cloud.storage.Storage.CopyRequest; import com.google.cloud.storage.StorageBatch; import com.google.cloud.storage.StorageBatchResult; @@ -705,9 +706,27 @@ public long bucketProject(GcsPath path) throws IOException { return bucket.getProject().longValue(); } - public void createBucket(BucketInfo bucketInfo) throws IOException { + public void createBucket(BucketInfo bucketInfo, BucketTargetOption... options) + throws IOException { + createBucket(null, bucketInfo, options); + } + + /** + * As {@link #createBucket(BucketInfo, BucketTargetOption...)}, but creates the bucket in the + * given project instead of the one this instance is configured with. + */ + public void createBucket( + @Nullable String projectId, BucketInfo bucketInfo, BucketTargetOption... options) + throws IOException { + Storage client = storage; + if (projectId != null && !projectId.equals(this.projectId)) { + // The owning project is a property of the client rather than of the insert request, so + // asking for a different one means deriving a client for it. The derived client shares the + // credentials, host and transport of the original. + client = storage.getOptions().toBuilder().setProjectId(projectId).build().getService(); + } try { - storage.create(bucketInfo); + client.create(bucketInfo, options); } catch (StorageException e) { throw translateStorageException(bucketInfo.getName(), null, e); } diff --git a/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilTest.java b/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilTest.java index 52bb877bb96c..e2a39ce06191 100644 --- a/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilTest.java +++ b/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilTest.java @@ -55,6 +55,7 @@ import com.google.api.client.testing.http.MockLowLevelHttpRequest; import com.google.api.client.testing.http.MockLowLevelHttpResponse; import com.google.api.client.util.BackOff; +import com.google.api.gax.paging.Page; import com.google.api.services.storage.Storage; import com.google.api.services.storage.model.Bucket; import com.google.api.services.storage.model.Objects; @@ -67,6 +68,9 @@ import com.google.cloud.hadoop.gcsio.GoogleCloudStorageOptions; import com.google.cloud.hadoop.gcsio.GoogleCloudStorageReadOptions; import com.google.cloud.hadoop.gcsio.StorageResourceId; +import com.google.cloud.storage.BucketInfo; +import com.google.cloud.storage.Storage.BucketTargetOption; +import com.google.cloud.storage.Storage.PredefinedAcl; import java.io.ByteArrayInputStream; import java.io.ByteArrayOutputStream; import java.io.FileNotFoundException; @@ -1887,4 +1891,200 @@ public void testGcsEndpoint() throws IOException { private static InputStream toStream(String content) throws IOException { return new ByteArrayInputStream(content.getBytes(StandardCharsets.UTF_8)); } + + // The tests below cover the routing that the GcsUtil facade performs for the legacy typed + // methods once the use_gcsutil_v2 experiment installs a GcsUtilV2 delegate. Both delegates are + // mocked, so they assert both that V2 is used and that V1 is left alone. + + private GcsUtilV1 mockDelegate; + private GcsUtilV2 mockDelegateV2; + + private GcsUtil gcsUtilRoutingToV2() { + GcsUtil gcsUtil = gcsOptionsWithTestCredential().getGcsUtil(); + mockDelegate = Mockito.mock(GcsUtilV1.class); + mockDelegateV2 = Mockito.mock(GcsUtilV2.class); + gcsUtil.delegate = mockDelegate; + gcsUtil.delegateV2 = mockDelegateV2; + return gcsUtil; + } + + private static com.google.cloud.storage.Blob mockBlob(String bucket, String object) { + com.google.cloud.storage.Blob blob = Mockito.mock(com.google.cloud.storage.Blob.class); + when(blob.getBucket()).thenReturn(bucket); + when(blob.getName()).thenReturn(object); + return blob; + } + + @Test + public void testCopyIsRoutedToV2AsAnUnconditionalOverwrite() throws IOException { + GcsUtil gcsUtil = gcsUtilRoutingToV2(); + + gcsUtil.copy(ImmutableList.of("gs://bucket/from"), ImmutableList.of("gs://bucket/to")); + + verify(mockDelegateV2) + .copy( + ImmutableList.of(GcsPath.fromUri("gs://bucket/from")), + ImmutableList.of(GcsPath.fromUri("gs://bucket/to")), + GcsUtilV2.OverwriteStrategy.ALWAYS_OVERWRITE); + Mockito.verifyNoMoreInteractions(mockDelegate); + } + + @Test + public void testRenameWithoutOptionsIsRoutedToV2AsAnUnconditionalOverwrite() throws IOException { + GcsUtil gcsUtil = gcsUtilRoutingToV2(); + + gcsUtil.rename(ImmutableList.of("gs://bucket/from"), ImmutableList.of("gs://bucket/to")); + + verify(mockDelegateV2) + .move( + ImmutableList.of(GcsPath.fromUri("gs://bucket/from")), + ImmutableList.of(GcsPath.fromUri("gs://bucket/to")), + GcsUtilV2.MissingStrategy.FAIL_IF_MISSING, + GcsUtilV2.OverwriteStrategy.ALWAYS_OVERWRITE); + Mockito.verifyNoMoreInteractions(mockDelegate); + } + + @Test + public void testRenameMoveOptionsAreTranslatedForV2() throws IOException { + GcsUtil gcsUtil = gcsUtilRoutingToV2(); + + gcsUtil.rename( + ImmutableList.of("gs://bucket/from"), + ImmutableList.of("gs://bucket/to"), + StandardMoveOptions.IGNORE_MISSING_FILES, + StandardMoveOptions.SKIP_IF_DESTINATION_EXISTS); + + verify(mockDelegateV2) + .move( + ImmutableList.of(GcsPath.fromUri("gs://bucket/from")), + ImmutableList.of(GcsPath.fromUri("gs://bucket/to")), + GcsUtilV2.MissingStrategy.SKIP_IF_MISSING, + GcsUtilV2.OverwriteStrategy.SKIP_IF_EXISTS); + Mockito.verifyNoMoreInteractions(mockDelegate); + } + + @Test + public void testRemoveIsRoutedToV2AndToleratesMissingFiles() throws IOException { + GcsUtil gcsUtil = gcsUtilRoutingToV2(); + + gcsUtil.remove(ImmutableList.of("gs://bucket/one", "gs://bucket/two")); + + verify(mockDelegateV2) + .remove( + ImmutableList.of( + GcsPath.fromUri("gs://bucket/one"), GcsPath.fromUri("gs://bucket/two")), + GcsUtilV2.MissingStrategy.SKIP_IF_MISSING); + Mockito.verifyNoMoreInteractions(mockDelegate); + } + + @Test + public void testGetObjectsIsRoutedToV2AndConvertsBlobs() throws IOException { + GcsUtil gcsUtil = gcsUtilRoutingToV2(); + com.google.cloud.storage.Blob blob = mockBlob("bucket", "found"); + when(blob.getSize()).thenReturn(42L); + when(blob.getMd5()).thenReturn("md5=="); + when(blob.getGeneration()).thenReturn(7L); + when(blob.getUpdateTimeOffsetDateTime()) + .thenReturn(java.time.Instant.ofEpochMilli(1234L).atOffset(java.time.ZoneOffset.UTC)); + FileNotFoundException notFound = new FileNotFoundException("gs://bucket/missing"); + List paths = + ImmutableList.of(GcsPath.fromUri("gs://bucket/found"), GcsPath.fromUri("gs://bucket/miss")); + when(mockDelegateV2.getBlobs(paths)) + .thenReturn( + ImmutableList.of( + GcsUtilV2.BlobResult.create(blob), GcsUtilV2.BlobResult.create(notFound))); + + List results = gcsUtil.getObjects(paths); + + assertEquals(2, results.size()); + StorageObject converted = results.get(0).storageObject(); + assertNotNull(converted); + assertEquals("bucket", converted.getBucket()); + assertEquals("found", converted.getName()); + assertEquals(BigInteger.valueOf(42L), converted.getSize()); + assertEquals("md5==", converted.getMd5Hash()); + assertEquals(Long.valueOf(7L), converted.getGeneration()); + assertEquals(1234L, converted.getUpdated().getValue()); + assertSame(notFound, results.get(1).ioException()); + assertNull(results.get(1).storageObject()); + Mockito.verifyNoMoreInteractions(mockDelegate); + } + + @Test + public void testListObjectsIsRoutedToV2AndConvertsAPage() throws IOException { + GcsUtil gcsUtil = gcsUtilRoutingToV2(); + com.google.cloud.storage.Blob object = mockBlob("bucket", "prefix/object"); + com.google.cloud.storage.Blob directory = mockBlob("bucket", "prefix/dir/"); + when(directory.isDirectory()).thenReturn(true); + @SuppressWarnings("unchecked") + Page page = Mockito.mock(Page.class); + when(page.getValues()).thenReturn(ImmutableList.of(object, directory)); + when(page.hasNextPage()).thenReturn(true); + when(page.getNextPageToken()).thenReturn("next"); + when(mockDelegateV2.listBlobs("bucket", "prefix/", null)).thenReturn(page); + + Objects objects = gcsUtil.listObjects("bucket", "prefix/", null); + + assertEquals(1, objects.getItems().size()); + assertEquals("prefix/object", objects.getItems().get(0).getName()); + assertEquals(ImmutableList.of("prefix/dir/"), objects.getPrefixes()); + assertEquals("next", objects.getNextPageToken()); + Mockito.verifyNoMoreInteractions(mockDelegate); + } + + @Test + public void testListObjectsReportsTheLastPageWithANullToken() throws IOException { + GcsUtil gcsUtil = gcsUtilRoutingToV2(); + @SuppressWarnings("unchecked") + Page page = Mockito.mock(Page.class); + when(page.getValues()).thenReturn(ImmutableList.of()); + // A gax page reports an empty token rather than a null one once it is exhausted. Callers of + // listObjects loop until the token is null, so it has to be normalized. + when(page.hasNextPage()).thenReturn(false); + when(page.getNextPageToken()).thenReturn(""); + when(mockDelegateV2.listBlobs("bucket", "prefix/", null)).thenReturn(page); + + Objects objects = gcsUtil.listObjects("bucket", "prefix/", null); + + assertNull(objects.getItems()); + assertNull(objects.getPrefixes()); + assertNull(objects.getNextPageToken()); + } + + @Test + public void testCreateBucketIsRoutedToV2WithProjectPrivateAcls() throws IOException { + GcsUtil gcsUtil = gcsUtilRoutingToV2(); + // This is the bucket that GcpOptions.tryCreateDefaultBucketWithPrefix builds. + Bucket bucket = + new Bucket() + .setName("bucket") + .setLocation("us-central1") + .setSoftDeletePolicy(new Bucket.SoftDeletePolicy().setRetentionDurationSeconds(0L)); + + gcsUtil.createBucket("a-project", bucket); + + verify(mockDelegateV2) + .createBucket( + "a-project", + BucketInfo.newBuilder("bucket") + .setLocation("us-central1") + .setSoftDeletePolicy( + BucketInfo.SoftDeletePolicy.newBuilder() + .setRetentionDuration(java.time.Duration.ZERO) + .build()) + .build(), + BucketTargetOption.predefinedAcl(PredefinedAcl.PROJECT_PRIVATE), + BucketTargetOption.predefinedDefaultObjectAcl(PredefinedAcl.PROJECT_PRIVATE)); + Mockito.verifyNoMoreInteractions(mockDelegate); + } + + @Test + public void testRemoveBucketIsRoutedToV2() throws IOException { + GcsUtil gcsUtil = gcsUtilRoutingToV2(); + + gcsUtil.removeBucket(new Bucket().setName("bucket")); + + verify(mockDelegateV2).removeBucket(BucketInfo.of("bucket")); + Mockito.verifyNoMoreInteractions(mockDelegate); + } } From d39af168793ec1c2f46c01a047081098ba2295ba Mon Sep 17 00:00:00 2001 From: Shunping Huang Date: Thu, 17 Sep 2026 13:57:55 -0400 Subject: [PATCH 2/9] Match GcsUtilV1's upload chunk size in GcsUtilV2 The formula is copied rather than read from AsyncWriteChannelOptions because GcsUtilV2 has no other gcsio reference and the migration ends with that dependency deleted. The test keeps the two values in sync. --- .../beam/sdk/extensions/gcp/util/GcsUtilV2.java | 13 ++++++++++--- .../beam/sdk/extensions/gcp/util/GcsUtilV2Test.java | 12 ++++++++++++ 2 files changed, 22 insertions(+), 3 deletions(-) diff --git a/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilV2.java b/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilV2.java index 8f72ced80f87..776bf6b47818 100644 --- a/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilV2.java +++ b/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilV2.java @@ -117,6 +117,14 @@ public GcsUtilV2 create(PipelineOptions options) { /** Maximum number of requests permitted in a GCS batch request. */ private static final int MAX_REQUESTS_PER_BATCH = 100; + /** + * Upload chunk size applied when the pipeline does not ask for one. Mirrors gcsio's {@code + * AsyncWriteChannelOptions} default, which java-storage does not share. + */ + @VisibleForTesting + static final int DEFAULT_UPLOAD_CHUNK_SIZE_BYTES = + Runtime.getRuntime().maxMemory() < 512 * 1024 * 1024 ? 8 * 1024 * 1024 : 3 * 8 * 1024 * 1024; + /** * Limit the number of bytes Cloud Storage will attempt to copy before responding to an individual * request. If you see Read Timeout errors, try reducing this value. @@ -892,9 +900,8 @@ public WritableByteChannel create( options.getUploadBufferSizeBytes() != null ? options.getUploadBufferSizeBytes() : this.uploadBufferSizeBytes; - if (uploadBufferSizeBytes != null) { - writer.setChunkSize(uploadBufferSizeBytes); - } + writer.setChunkSize( + uploadBufferSizeBytes != null ? uploadBufferSizeBytes : DEFAULT_UPLOAD_CHUNK_SIZE_BYTES); serviceCallMetric.call("ok"); // Return the bridge wrapper diff --git a/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilV2Test.java b/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilV2Test.java index 2b9e55b98b86..8bd79860cc28 100644 --- a/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilV2Test.java +++ b/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilV2Test.java @@ -30,6 +30,7 @@ import com.google.api.client.testing.http.MockLowLevelHttpResponse; import com.google.auth.Credentials; import com.google.cloud.NoCredentials; +import com.google.cloud.hadoop.util.AsyncWriteChannelOptions; import com.google.cloud.http.HttpTransportOptions; import com.google.cloud.storage.Storage; import com.google.cloud.storage.StorageOptions; @@ -415,4 +416,15 @@ public void testWireBytesAreAttributedToTheBoundContainer() throws IOException { assertNull(gcsCounter(other, "gcs_http_read_wire_bytes_received")); assertNull(gcsCounter(other, "gcs_http_write_wire_bytes_sent")); } + + /** + * The chunk size decides how many requests a write costs, and the two clients do not default to + * the same one, so a drift here is a silent throughput regression rather than a test failure. + */ + @Test + public void testDefaultUploadChunkSizeMatchesV1() { + assertEquals( + AsyncWriteChannelOptions.DEFAULT.getUploadChunkSize(), + GcsUtilV2.DEFAULT_UPLOAD_CHUNK_SIZE_BYTES); + } } From 4a00cd4321a29dbac9944256b16d36947c8ce7d0 Mon Sep 17 00:00:00 2001 From: Shunping Huang Date: Wed, 23 Sep 2026 12:49:22 -0400 Subject: [PATCH 3/9] Fix a bug of use_gcsutil_v2 flag not passing to experiments in ParquetIOLT --- .../beam/it/gcp/storage/ParquetIOLT.java | 47 ++++++++++++------- 1 file changed, 29 insertions(+), 18 deletions(-) diff --git a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/storage/ParquetIOLT.java b/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/storage/ParquetIOLT.java index 80eb7886c95e..52b12372dcd6 100644 --- a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/storage/ParquetIOLT.java +++ b/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/storage/ParquetIOLT.java @@ -31,7 +31,9 @@ import java.time.Duration; import java.time.ZoneOffset; import java.time.format.DateTimeFormatter; +import java.util.ArrayList; import java.util.Arrays; +import java.util.List; import java.util.Map; import java.util.Random; import java.util.UUID; @@ -120,9 +122,9 @@ * *

Whichever is chosen, the job is launched with an explicit experiment, {@code use_runner_v2} or * {@code disable_runner_v2}. Leaving the choice to the service is not an option here because the - * container image is resolved on the client, see {@link #dataflowWorkerExperiment()}: a job - * submitted without an experiment ends up asking for an image tag that does not exist and hangs - * with the workers in ImagePullBackOff. + * container image is resolved on the client, see {@link #dataflowExperiments()}: a job submitted + * without an experiment ends up asking for an image tag that does not exist and hangs with the + * workers in ImagePullBackOff. * *

Note that Runner v2 stages the locally built SDK jars, so a local SDK change is measured as * is, while the legacy worker runs the Beam code baked into its container image. @@ -198,7 +200,7 @@ public final class ParquetIOLT extends GcsIOLoadTestBase { /** * Experiment that keeps the job on the legacy worker. Needed even though the legacy worker is * what a job without experiments is submitted as, because the service upgrades such a job to - * Runner v2 on its own, see {@link #dataflowWorkerExperiment()}. + * Runner v2 on its own, see {@link #dataflowExperiments()}. */ private static final String LEGACY_WORKER_EXPERIMENT = "disable_runner_v2"; @@ -499,8 +501,8 @@ private PipelineLauncher.LaunchConfig launchConfig(String jobName, TestPipeline // the gcs_* metrics measure. // maxNumWorkers is deliberately not set, it only bounds an autoscaling pool. builder - // Picks the worker, see dataflowWorkerExperiment(). - .addParameter("experiments", dataflowWorkerExperiment()) + // Picks the worker and the GcsUtil version, see dataflowExperiments(). + .addParameter("experiments", dataflowExperiments()) .addParameter("autoscalingAlgorithm", "NONE") .addParameter("numWorkers", String.valueOf(DATAFLOW_NUM_WORKERS)) .addParameter("workerMachineType", DATAFLOW_MACHINE_TYPE); @@ -510,14 +512,19 @@ private PipelineLauncher.LaunchConfig launchConfig(String jobName, TestPipeline } /** - * Experiment that selects the Dataflow worker, {@code use_runner_v2} or {@code - * disable_runner_v2}. + * Every experiment the Dataflow job is submitted with, as the comma separated list the {@code + * experiments} pipeline option parses. * - *

The worker is always selected explicitly, even though Runner v2 is what the service picks on - * its own, because the container image is resolved on the client: {@code - * DataflowRunner.getDefaultContainerImageUrl} takes the image name and the image tag from the - * same branch of its {@code useUnifiedWorker()} check, and the two tags ({@code - * dataflowFnapiContainerVersion} and {@code dataflowLegacyContainerVersion} in {@code + *

They have to be joined into one value because the launcher carries the parameters in a map, + * so {@code experiments} can only be given once. That map is also the only channel that reaches + * the job: the launcher rebuilds the pipeline options from these parameters alone, so an + * experiment set on the options of the pipeline would be dropped. + * + *

The list always names the worker, {@code use_runner_v2} or {@code disable_runner_v2}, even + * though Runner v2 is what the service picks on its own, because the container image is resolved + * on the client: {@code DataflowRunner.getDefaultContainerImageUrl} takes the image name and the + * image tag from the same branch of its {@code useUnifiedWorker()} check, and the two tags + * ({@code dataflowFnapiContainerVersion} and {@code dataflowLegacyContainerVersion} in {@code * runners/google-cloud-dataflow-java/build.gradle}) are bumped independently. * *

A job submitted without an experiment therefore resolves the legacy pair {@code @@ -529,8 +536,13 @@ private PipelineLauncher.LaunchConfig launchConfig(String jobName, TestPipeline * gives {@code beam_javaNN_sdk:}, {@code disable_runner_v2} keeps the service from * upgrading the job so {@code beam-javaNN-batch:} stays correct. */ - private static String dataflowWorkerExperiment() { - return configuration.useRunnerV2 ? RUNNER_V2_EXPERIMENT : LEGACY_WORKER_EXPERIMENT; + private static String dataflowExperiments() { + List experiments = new ArrayList<>(); + experiments.add(configuration.useRunnerV2 ? RUNNER_V2_EXPERIMENT : LEGACY_WORKER_EXPERIMENT); + if (configuration.useGcsUtilV2) { + experiments.add(GCS_UTIL_V2_EXPERIMENT); + } + return String.join(",", experiments); } /** @@ -669,7 +681,7 @@ private static void printConfiguration() { DATAFLOW_RUNNER.equalsIgnoreCase(configuration.runner) ? String.format( "%s (--experiments=%s)", - configuration.useRunnerV2 ? "Runner v2" : "legacy", dataflowWorkerExperiment()) + configuration.useRunnerV2 ? "Runner v2" : "legacy", dataflowExperiments()) : "n/a", DATAFLOW_RUNNER.equalsIgnoreCase(configuration.runner) ? String.format("%d x %s, autoscaling off", DATAFLOW_NUM_WORKERS, DATAFLOW_MACHINE_TYPE) @@ -875,8 +887,7 @@ static class Configuration extends SyntheticSourceOptions { /** * Dataflow only. {@code true} runs the job on Runner v2, i.e. the unified worker, {@code false} * on the legacy worker. Either way the choice is sent to the service as an explicit experiment, - * see {@link ParquetIOLT#dataflowWorkerExperiment()} for why it must not be left to the - * service. + * see {@link ParquetIOLT#dataflowExperiments()} for why it must not be left to the service. */ @JsonProperty public boolean useRunnerV2 = true; From f29db31d20598c44dc456d71bed0d89bf6e999a5 Mon Sep 17 00:00:00 2001 From: Shunping Huang Date: Thu, 24 Sep 2026 23:56:22 -0400 Subject: [PATCH 4/9] Fix GcsUtilV2 parity gaps. - Translate StorageException thrown on write channel close. - Record a 404 request metric when open() hits a missing object. --- .../sdk/extensions/gcp/util/GcsUtilV2.java | 12 ++- .../sdk/extensions/gcp/util/GcsUtilTest.java | 80 +++++++++++++++++++ 2 files changed, 91 insertions(+), 1 deletion(-) diff --git a/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilV2.java b/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilV2.java index 776bf6b47818..a0917724b474 100644 --- a/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilV2.java +++ b/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilV2.java @@ -828,6 +828,11 @@ public SeekableByteChannel open(GcsPath path, BlobSourceOption... sourceOptions) serviceCallMetric.call("ok"); return wrapInCounting( new GcsSeekableByteChannel(reader, blob.getSize()), path.getBucket(), container); + } catch (FileNotFoundException e) { + // getBlob reports a missing object as a FileNotFoundException rather than a + // StorageException, so record its status here like GcsUtilV1 does. + serviceCallMetric.call(404); + throw e; } catch (StorageException e) { serviceCallMetric.call(e.getCode()); throw translateStorageException(path, e); @@ -860,7 +865,12 @@ public boolean isOpen() { @Override public void close() throws IOException { - writer.close(); + // The upload is finalized here, so this is where a failed precondition surfaces. + try { + writer.close(); + } catch (StorageException e) { + throw translateStorageException(gcsPath, e); + } } } diff --git a/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilTest.java b/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilTest.java index e2a39ce06191..448159a05f90 100644 --- a/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilTest.java +++ b/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilTest.java @@ -30,6 +30,7 @@ import static org.junit.Assert.assertThrows; import static org.junit.Assert.assertTrue; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -62,6 +63,7 @@ import com.google.api.services.storage.model.RewriteResponse; import com.google.api.services.storage.model.StorageObject; import com.google.auth.Credentials; +import com.google.cloud.WriteChannel; import com.google.cloud.hadoop.gcsio.CreateObjectOptions; import com.google.cloud.hadoop.gcsio.GoogleCloudStorage; import com.google.cloud.hadoop.gcsio.GoogleCloudStorageImpl; @@ -71,6 +73,7 @@ import com.google.cloud.storage.BucketInfo; import com.google.cloud.storage.Storage.BucketTargetOption; import com.google.cloud.storage.Storage.PredefinedAcl; +import com.google.cloud.storage.StorageException; import java.io.ByteArrayInputStream; import java.io.ByteArrayOutputStream; import java.io.FileNotFoundException; @@ -2087,4 +2090,81 @@ public void testRemoveBucketIsRoutedToV2() throws IOException { verify(mockDelegateV2).removeBucket(BucketInfo.of("bucket")); Mockito.verifyNoMoreInteractions(mockDelegate); } + + // The tests below exercise a real GcsUtilV2 delegate whose java-storage client is mocked, to + // cover behavior that GcsUtilV2 must share with GcsUtilV1. + // TODO: Move these to a parity test that runs against both delegates. + + /** Returns a {@link GcsUtil} backed by a real {@link GcsUtilV2} that calls {@code storage}. */ + private GcsUtil gcsUtilWithV2Storage(com.google.cloud.storage.Storage storage) { + GcsOptions options = gcsOptionsWithTestCredential(); + options.setProject("my_project"); + GcsUtil gcsUtil = options.getGcsUtil(); + GcsUtilV2 delegateV2 = Mockito.spy(new GcsUtilV2(options)); + Mockito.doReturn(storage).when(delegateV2).storageWithHttpMetrics(any(), anyBoolean()); + gcsUtil.delegateV2 = delegateV2; + return gcsUtil; + } + + /** + * Mirrors {@link #testGCSReadMetricsIsSet} for V2: opening a missing object records a {@code + * not_found} request, even though V2 detects it through a lookup rather than a {@link + * StorageException}. + */ + @Test + public void testV2OpenMissingObjectRecordsNotFoundMetric() { + com.google.cloud.storage.Storage storage = Mockito.mock(com.google.cloud.storage.Storage.class); + // An unstubbed get() returns null, which is how java-storage reports a missing object. + GcsUtil gcsUtil = gcsUtilWithV2Storage(storage); + + assertThrows( + FileNotFoundException.class, + () -> gcsUtil.open(GcsPath.fromComponents("testbucket", "testobject"))); + + verifyMetricWasSet("my_project", "testbucket", "GcsGet", "not_found", 1); + verifyMetricWasSet("my_project", "testbucket", "GcsGet", "ok", 0); + } + + /** + * An upload is finalized when its channel is closed, so a failed precondition surfaces there. V2 + * must report it as an {@link IOException}, as V1 does, rather than an unchecked {@link + * StorageException}. + */ + @Test + public void testV2WriteChannelCloseTranslatesStorageException() throws IOException { + com.google.cloud.storage.Storage storage = Mockito.mock(com.google.cloud.storage.Storage.class); + WriteChannel writer = Mockito.mock(WriteChannel.class); + StorageException preconditionFailed = new StorageException(412, "Precondition Failed"); + Mockito.doThrow(preconditionFailed).when(writer).close(); + when(storage.writer(any(com.google.cloud.storage.BlobInfo.class), any())).thenReturn(writer); + GcsUtil gcsUtil = gcsUtilWithV2Storage(storage); + + WritableByteChannel channel = + gcsUtil.create( + GcsPath.fromComponents("testbucket", "testobject"), + CreateOptions.builder().setExpectFileToNotExist(true).build()); + + IOException thrown = assertThrows(IOException.class, channel::close); + assertSame(preconditionFailed, thrown.getCause()); + } + + /** Well-known failures on close are translated like any other V2 call, e.g. 403 and 404. */ + @Test + public void testV2WriteChannelCloseTranslatesKnownStatusCodes() throws IOException { + com.google.cloud.storage.Storage storage = Mockito.mock(com.google.cloud.storage.Storage.class); + WriteChannel writer = Mockito.mock(WriteChannel.class); + Mockito.doThrow(new StorageException(403, "Forbidden")) + .doThrow(new StorageException(404, "Not Found")) + .when(writer) + .close(); + when(storage.writer(any(com.google.cloud.storage.BlobInfo.class), any())).thenReturn(writer); + GcsUtil gcsUtil = gcsUtilWithV2Storage(storage); + GcsPath path = GcsPath.fromComponents("testbucket", "testobject"); + CreateOptions options = CreateOptions.builder().setExpectFileToNotExist(true).build(); + + WritableByteChannel forbidden = gcsUtil.create(path, options); + assertThrows(AccessDeniedException.class, forbidden::close); + WritableByteChannel notFound = gcsUtil.create(path, options); + assertThrows(FileNotFoundException.class, notFound::close); + } } From 3795d091bb447b4815671a4d6ca76247783d59ee Mon Sep 17 00:00:00 2001 From: Shunping Huang Date: Fri, 25 Sep 2026 09:41:02 -0400 Subject: [PATCH 5/9] Fix bucketAccessible in V2 to match V1 --- .../sdk/extensions/gcp/util/GcsUtilV2.java | 10 +++-- .../sdk/extensions/gcp/util/GcsUtilTest.java | 45 +++++++++++++++++++ 2 files changed, 52 insertions(+), 3 deletions(-) diff --git a/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilV2.java b/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilV2.java index a0917724b474..15c3f728dacb 100644 --- a/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilV2.java +++ b/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilV2.java @@ -684,13 +684,17 @@ public Bucket getBucket(GcsPath path, BucketGetOption... options) throws IOExcep } } - /** Returns whether the GCS bucket exists and is accessible. */ - public boolean bucketAccessible(GcsPath path) { + /** + * Returns whether the GCS bucket exists and is accessible. This will return false if the bucket + * does not exist or is inaccessible due to permissions; any other failure is propagated, as + * {@link GcsUtilV1} does. + */ + public boolean bucketAccessible(GcsPath path) throws IOException { try { // Fetch only the name field to minimize data transfer getBucket(path, BucketGetOption.fields(BucketField.NAME)); return true; - } catch (IOException e) { + } catch (AccessDeniedException | FileNotFoundException e) { return false; } } diff --git a/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilTest.java b/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilTest.java index 448159a05f90..cea43d7dd786 100644 --- a/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilTest.java +++ b/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilTest.java @@ -2167,4 +2167,49 @@ public void testV2WriteChannelCloseTranslatesKnownStatusCodes() throws IOExcepti WritableByteChannel notFound = gcsUtil.create(path, options); assertThrows(FileNotFoundException.class, notFound::close); } + + /** Returns a {@link GcsUtil} backed by a real {@link GcsUtilV2} whose bucket lookup throws. */ + private GcsUtil gcsUtilWithV2BucketLookupFailing(IOException failure) throws IOException { + GcsOptions options = gcsOptionsWithTestCredential(); + options.setProject("my_project"); + GcsUtil gcsUtil = options.getGcsUtil(); + GcsUtilV2 delegateV2 = Mockito.spy(new GcsUtilV2(options)); + Mockito.doThrow(failure).when(delegateV2).getBucket(any(GcsPath.class), any()); + gcsUtil.delegateV2 = delegateV2; + return gcsUtil; + } + + /** Mirrors {@link #testBucketDoesNotExist} for V2. */ + @Test + public void testV2BucketAccessibleIsFalseWhenBucketDoesNotExist() throws IOException { + GcsUtil gcsUtil = + gcsUtilWithV2BucketLookupFailing(new FileNotFoundException("gs://testbucket")); + + assertFalse(gcsUtil.bucketAccessible(GcsPath.fromComponents("testbucket", "testobject"))); + } + + /** Mirrors {@link #testBucketDoesNotExistBecauseOfAccessError} for V2. */ + @Test + public void testV2BucketAccessibleIsFalseWhenAccessIsDenied() throws IOException { + GcsUtil gcsUtil = + gcsUtilWithV2BucketLookupFailing(new AccessDeniedException("gs://testbucket")); + + assertFalse(gcsUtil.bucketAccessible(GcsPath.fromComponents("testbucket", "testobject"))); + } + + /** + * Any other failure (e.g. a 5xx) says nothing about whether the bucket is accessible, so it must + * propagate rather than be reported as an inaccessible bucket, as V1 does. + */ + @Test + public void testV2BucketAccessiblePropagatesOtherFailures() throws IOException { + IOException serverError = new IOException(new StorageException(503, "Service Unavailable")); + GcsUtil gcsUtil = gcsUtilWithV2BucketLookupFailing(serverError); + + IOException thrown = + assertThrows( + IOException.class, + () -> gcsUtil.bucketAccessible(GcsPath.fromComponents("testbucket", "testobject"))); + assertSame(serverError, thrown); + } } From 9c107663bbc2495ad53ecead67c03aca640979c2 Mon Sep 17 00:00:00 2001 From: Shunping Huang Date: Fri, 25 Sep 2026 00:40:13 -0400 Subject: [PATCH 6/9] Route the deprecated getBucket and create(path, type) through GcsUtilV2 --- .../beam/sdk/extensions/gcp/util/GcsUtil.java | 40 +++- .../sdk/extensions/gcp/util/GcsUtilTest.java | 186 ++++++++++++++++++ 2 files changed, 224 insertions(+), 2 deletions(-) diff --git a/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtil.java b/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtil.java index 908981d3bc87..382ac6856bf4 100644 --- a/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtil.java +++ b/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtil.java @@ -272,7 +272,9 @@ public SeekableByteChannel openV2(GcsPath path, BlobSourceOption... options) thr */ @Deprecated public WritableByteChannel create(GcsPath path, String type) throws IOException { - return delegate.create(path, type); + // Built the same way as GcsUtilV1#create(GcsPath, String), but through this class so that it + // follows GcsUtilV2 when enabled. + return create(path, CreateOptions.builder().setContentType(type).build()); } /** @@ -281,7 +283,12 @@ public WritableByteChannel create(GcsPath path, String type) throws IOException @Deprecated public WritableByteChannel create(GcsPath path, String type, Integer uploadBufferSizeBytes) throws IOException { - return delegate.create(path, type, uploadBufferSizeBytes); + return create( + path, + CreateOptions.builder() + .setContentType(type) + .setUploadBufferSizeBytes(uploadBufferSizeBytes) + .build()); } public static class CreateOptions { @@ -404,6 +411,9 @@ public void createBucket(BucketInfo bucketInfo, BucketTargetOption... options) */ @Deprecated public @Nullable Bucket getBucket(GcsPath path) throws IOException { + if (delegateV2 != null) { + return toBucket(delegateV2.getBucket(path)); + } return delegate.getBucket(path); } @@ -580,6 +590,32 @@ private static BucketInfo toBucketInfo(Bucket bucket) { return builder.build(); } + /** + * Converts a java-storage {@link com.google.cloud.storage.Bucket} back into the JSON API {@link + * Bucket} model, for the deprecated {@link #getBucket(GcsPath)}. + * + *

The reverse of {@link #toBucketInfo}, plus the owning project number. Like it, this is + * expected to go away with the deprecated methods it serves. + */ + private static Bucket toBucket(com.google.cloud.storage.Bucket bucketInfo) { + Bucket bucket = + new Bucket() + .setName(bucketInfo.getName()) + .setLocation(bucketInfo.getLocation()) + .setProjectNumber(bucketInfo.getProject()); + StorageClass storageClass = bucketInfo.getStorageClass(); + if (storageClass != null) { + bucket.setStorageClass(storageClass.name()); + } + BucketInfo.SoftDeletePolicy softDeletePolicy = bucketInfo.getSoftDeletePolicy(); + Duration retention = softDeletePolicy == null ? null : softDeletePolicy.getRetentionDuration(); + if (retention != null) { + bucket.setSoftDeletePolicy( + new Bucket.SoftDeletePolicy().setRetentionDurationSeconds(retention.getSeconds())); + } + return bucket; + } + /** * Converts a java-storage {@link Blob} back into the JSON API {@link StorageObject} model. * diff --git a/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilTest.java b/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilTest.java index cea43d7dd786..ebe550dd329e 100644 --- a/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilTest.java +++ b/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilTest.java @@ -73,6 +73,7 @@ import com.google.cloud.storage.BucketInfo; import com.google.cloud.storage.Storage.BucketTargetOption; import com.google.cloud.storage.Storage.PredefinedAcl; +import com.google.cloud.storage.StorageClass; import com.google.cloud.storage.StorageException; import java.io.ByteArrayInputStream; import java.io.ByteArrayOutputStream; @@ -124,6 +125,7 @@ import org.junit.Before; import org.junit.Rule; import org.junit.Test; +import org.junit.function.ThrowingRunnable; import org.junit.rules.ExpectedException; import org.junit.runner.RunWith; import org.junit.runners.JUnit4; @@ -2091,6 +2093,190 @@ public void testRemoveBucketIsRoutedToV2() throws IOException { Mockito.verifyNoMoreInteractions(mockDelegate); } + @Test + public void testBucketOwnerIsRoutedToV2BucketProject() throws IOException { + GcsUtil gcsUtil = gcsUtilRoutingToV2(); + GcsPath path = GcsPath.fromUri("gs://bucket/object"); + when(mockDelegateV2.bucketProject(path)).thenReturn(123L); + + assertEquals(123L, gcsUtil.bucketOwner(path)); + Mockito.verifyNoMoreInteractions(mockDelegate); + } + + /** The emulator ignores the storage class, so only a unit test can see it carried over. */ + @Test + public void testCreateBucketKeepsTheStorageClassForV2() throws IOException { + GcsUtil gcsUtil = gcsUtilRoutingToV2(); + + gcsUtil.createBucket("a-project", new Bucket().setName("bucket").setStorageClass("NEARLINE")); + + verify(mockDelegateV2) + .createBucket( + "a-project", + BucketInfo.newBuilder("bucket").setStorageClass(StorageClass.NEARLINE).build(), + BucketTargetOption.predefinedAcl(PredefinedAcl.PROJECT_PRIVATE), + BucketTargetOption.predefinedDefaultObjectAcl(PredefinedAcl.PROJECT_PRIVATE)); + } + + @Test + public void testGetObjectIsRoutedToV2AndKeepsAllFields() throws IOException { + GcsUtil gcsUtil = gcsUtilRoutingToV2(); + GcsPath path = GcsPath.fromUri("gs://bucket/object"); + com.google.cloud.storage.Blob blob = mockBlob("bucket", "object"); + when(blob.getSize()).thenReturn(42L); + when(blob.getGeneration()).thenReturn(7L); + when(blob.getMetageneration()).thenReturn(3L); + when(blob.getContentType()).thenReturn("text/csv"); + when(blob.getContentEncoding()).thenReturn("gzip"); + when(blob.getMd5()).thenReturn("md5=="); + when(blob.getCrc32c()).thenReturn("crc=="); + when(blob.getEtag()).thenReturn("etag"); + when(blob.getUpdateTimeOffsetDateTime()) + .thenReturn(java.time.Instant.ofEpochMilli(1234L).atOffset(java.time.ZoneOffset.UTC)); + when(blob.getCreateTimeOffsetDateTime()) + .thenReturn(java.time.Instant.ofEpochMilli(1000L).atOffset(java.time.ZoneOffset.UTC)); + when(mockDelegateV2.getBlob(path)).thenReturn(blob); + + StorageObject object = gcsUtil.getObject(path); + + assertEquals("bucket", object.getBucket()); + assertEquals("object", object.getName()); + assertEquals(BigInteger.valueOf(42L), object.getSize()); + assertEquals(Long.valueOf(7L), object.getGeneration()); + assertEquals(Long.valueOf(3L), object.getMetageneration()); + assertEquals("text/csv", object.getContentType()); + assertEquals("gzip", object.getContentEncoding()); + assertEquals("md5==", object.getMd5Hash()); + assertEquals("crc==", object.getCrc32c()); + assertEquals("etag", object.getEtag()); + assertEquals(1234L, object.getUpdated().getValue()); + assertEquals(1000L, object.getTimeCreated().getValue()); + Mockito.verifyNoMoreInteractions(mockDelegate); + } + + /** Fields a blob may not carry (e.g. when fetched with a field mask) are left unset. */ + @Test + public void testGetObjectLeavesMissingFieldsUnsetForV2() throws IOException { + GcsUtil gcsUtil = gcsUtilRoutingToV2(); + GcsPath path = GcsPath.fromUri("gs://bucket/object"); + com.google.cloud.storage.Blob blob = mockBlob("bucket", "object"); + // Mockito would otherwise answer 0 for the boxed size. + when(blob.getSize()).thenReturn(null); + when(mockDelegateV2.getBlob(path)).thenReturn(blob); + + StorageObject object = gcsUtil.getObject(path); + + assertEquals("object", object.getName()); + assertNull(object.getSize()); + assertNull(object.getUpdated()); + assertNull(object.getTimeCreated()); + } + + @Test + public void testCreateWithTypeIsRoutedToV2() throws IOException { + GcsUtil gcsUtil = gcsUtilRoutingToV2(); + GcsPath path = GcsPath.fromUri("gs://bucket/object"); + + gcsUtil.create(path, "text/plain"); + gcsUtil.create(path, "text/plain", 1024); + + verify(mockDelegateV2) + .create(path, GcsUtilV1.CreateOptions.builder().setContentType("text/plain").build()); + verify(mockDelegateV2) + .create( + path, + GcsUtilV1.CreateOptions.builder() + .setContentType("text/plain") + .setUploadBufferSizeBytes(1024) + .build()); + Mockito.verifyNoMoreInteractions(mockDelegate); + } + + /** GcpOptions reads the soft delete policy of the temp bucket through this method. */ + @Test + public void testGetBucketIsRoutedToV2AndConvertsTheBucket() throws IOException { + GcsUtil gcsUtil = gcsUtilRoutingToV2(); + GcsPath path = GcsPath.fromUri("gs://bucket/object"); + com.google.cloud.storage.Bucket bucket = Mockito.mock(com.google.cloud.storage.Bucket.class); + when(bucket.getName()).thenReturn("bucket"); + when(bucket.getLocation()).thenReturn("US-CENTRAL1"); + when(bucket.getProject()).thenReturn(BigInteger.valueOf(123L)); + when(bucket.getStorageClass()).thenReturn(StorageClass.NEARLINE); + when(bucket.getSoftDeletePolicy()) + .thenReturn( + BucketInfo.SoftDeletePolicy.newBuilder() + .setRetentionDuration(java.time.Duration.ofDays(7)) + .build()); + when(mockDelegateV2.getBucket(path)).thenReturn(bucket); + + Bucket converted = gcsUtil.getBucket(path); + + assertNotNull(converted); + assertEquals("bucket", converted.getName()); + assertEquals("US-CENTRAL1", converted.getLocation()); + assertEquals(BigInteger.valueOf(123L), converted.getProjectNumber()); + assertEquals("NEARLINE", converted.getStorageClass()); + assertEquals( + Long.valueOf(java.time.Duration.ofDays(7).getSeconds()), + converted.getSoftDeletePolicy().getRetentionDurationSeconds()); + Mockito.verifyNoMoreInteractions(mockDelegate); + } + + @Test + public void testGetBucketWithoutSoftDeletePolicyForV2() throws IOException { + GcsUtil gcsUtil = gcsUtilRoutingToV2(); + GcsPath path = GcsPath.fromUri("gs://bucket/object"); + com.google.cloud.storage.Bucket bucket = Mockito.mock(com.google.cloud.storage.Bucket.class); + when(bucket.getName()).thenReturn("bucket"); + when(mockDelegateV2.getBucket(path)).thenReturn(bucket); + + Bucket converted = gcsUtil.getBucket(path); + + assertNotNull(converted); + assertNull(converted.getSoftDeletePolicy()); + assertNull(converted.getStorageClass()); + } + + /** Without the use_gcsutil_v2 experiment, the V2-only methods fail rather than fall back. */ + @Test + public void testV2OnlyMethodsFailWithoutV2() { + GcsUtil gcsUtil = gcsOptionsWithTestCredential().getGcsUtil(); + assertNull(gcsUtil.delegateV2); + GcsUtilV1 v1 = Mockito.mock(GcsUtilV1.class); + gcsUtil.delegate = v1; + GcsPath path = GcsPath.fromUri("gs://bucket/object"); + List paths = ImmutableList.of(path); + + List calls = + ImmutableList.of( + () -> gcsUtil.getBlob(path), + () -> gcsUtil.getBlobs(paths), + () -> gcsUtil.listBlobs("bucket", "prefix", null), + () -> gcsUtil.listBlobs("bucket", "prefix", null, "/"), + () -> gcsUtil.openV2(path), + () -> gcsUtil.createV2(path, GcsUtil.CreateOptions.builder().build()), + () -> gcsUtil.createBucket(BucketInfo.of("bucket")), + () -> gcsUtil.getBucketWithOptions(path), + () -> gcsUtil.removeBucket(BucketInfo.of("bucket")), + () -> gcsUtil.copyV2(paths, paths), + () -> gcsUtil.copy(paths, paths, GcsUtilV2.OverwriteStrategy.ALWAYS_OVERWRITE), + () -> gcsUtil.renameV2(paths, paths), + () -> + gcsUtil.rename( + paths, + paths, + GcsUtilV2.MissingStrategy.FAIL_IF_MISSING, + GcsUtilV2.OverwriteStrategy.ALWAYS_OVERWRITE), + () -> gcsUtil.removeV2(paths), + () -> gcsUtil.remove(paths, GcsUtilV2.MissingStrategy.FAIL_IF_MISSING)); + + for (ThrowingRunnable call : calls) { + IOException e = assertThrows(IOException.class, call); + assertEquals("GcsUtil V2 not initialized.", e.getMessage()); + } + Mockito.verifyNoInteractions(v1); + } + // The tests below exercise a real GcsUtilV2 delegate whose java-storage client is mocked, to // cover behavior that GcsUtilV2 must share with GcsUtilV1. // TODO: Move these to a parity test that runs against both delegates. From 41d5676b7e437d4a6fd174706474f65c2c733270 Mon Sep 17 00:00:00 2001 From: Shunping Huang Date: Fri, 25 Sep 2026 09:50:55 -0400 Subject: [PATCH 7/9] Add some more routing tests and trim unnecessary ones. --- .../sdk/extensions/gcp/util/GcsUtilTest.java | 134 ++++++++++++------ 1 file changed, 88 insertions(+), 46 deletions(-) diff --git a/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilTest.java b/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilTest.java index ebe550dd329e..eceaa15b7bd7 100644 --- a/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilTest.java +++ b/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilTest.java @@ -1982,15 +1982,14 @@ public void testRemoveIsRoutedToV2AndToleratesMissingFiles() throws IOException Mockito.verifyNoMoreInteractions(mockDelegate); } + /** + * Blobs and errors are passed through in order. The field-by-field conversion is covered by + * {@link #testGetObjectIsRoutedToV2AndKeepsAllFields}. + */ @Test public void testGetObjectsIsRoutedToV2AndConvertsBlobs() throws IOException { GcsUtil gcsUtil = gcsUtilRoutingToV2(); com.google.cloud.storage.Blob blob = mockBlob("bucket", "found"); - when(blob.getSize()).thenReturn(42L); - when(blob.getMd5()).thenReturn("md5=="); - when(blob.getGeneration()).thenReturn(7L); - when(blob.getUpdateTimeOffsetDateTime()) - .thenReturn(java.time.Instant.ofEpochMilli(1234L).atOffset(java.time.ZoneOffset.UTC)); FileNotFoundException notFound = new FileNotFoundException("gs://bucket/missing"); List paths = ImmutableList.of(GcsPath.fromUri("gs://bucket/found"), GcsPath.fromUri("gs://bucket/miss")); @@ -2004,12 +2003,8 @@ public void testGetObjectsIsRoutedToV2AndConvertsBlobs() throws IOException { assertEquals(2, results.size()); StorageObject converted = results.get(0).storageObject(); assertNotNull(converted); - assertEquals("bucket", converted.getBucket()); assertEquals("found", converted.getName()); - assertEquals(BigInteger.valueOf(42L), converted.getSize()); - assertEquals("md5==", converted.getMd5Hash()); - assertEquals(Long.valueOf(7L), converted.getGeneration()); - assertEquals(1234L, converted.getUpdated().getValue()); + assertNull(results.get(0).ioException()); assertSame(notFound, results.get(1).ioException()); assertNull(results.get(1).storageObject()); Mockito.verifyNoMoreInteractions(mockDelegate); @@ -2056,14 +2051,20 @@ public void testListObjectsReportsTheLastPageWithANullToken() throws IOException assertNull(objects.getNextPageToken()); } + /** + * The storage class is checked here because the emulator ignores it, so only a unit test can see + * it carried over. + */ @Test public void testCreateBucketIsRoutedToV2WithProjectPrivateAcls() throws IOException { GcsUtil gcsUtil = gcsUtilRoutingToV2(); - // This is the bucket that GcpOptions.tryCreateDefaultBucketWithPrefix builds. + // This is the bucket that GcpOptions.tryCreateDefaultBucketWithPrefix builds, plus a storage + // class. Bucket bucket = new Bucket() .setName("bucket") .setLocation("us-central1") + .setStorageClass("NEARLINE") .setSoftDeletePolicy(new Bucket.SoftDeletePolicy().setRetentionDurationSeconds(0L)); gcsUtil.createBucket("a-project", bucket); @@ -2073,6 +2074,7 @@ public void testCreateBucketIsRoutedToV2WithProjectPrivateAcls() throws IOExcept "a-project", BucketInfo.newBuilder("bucket") .setLocation("us-central1") + .setStorageClass(StorageClass.NEARLINE) .setSoftDeletePolicy( BucketInfo.SoftDeletePolicy.newBuilder() .setRetentionDuration(java.time.Duration.ZERO) @@ -2103,21 +2105,6 @@ public void testBucketOwnerIsRoutedToV2BucketProject() throws IOException { Mockito.verifyNoMoreInteractions(mockDelegate); } - /** The emulator ignores the storage class, so only a unit test can see it carried over. */ - @Test - public void testCreateBucketKeepsTheStorageClassForV2() throws IOException { - GcsUtil gcsUtil = gcsUtilRoutingToV2(); - - gcsUtil.createBucket("a-project", new Bucket().setName("bucket").setStorageClass("NEARLINE")); - - verify(mockDelegateV2) - .createBucket( - "a-project", - BucketInfo.newBuilder("bucket").setStorageClass(StorageClass.NEARLINE).build(), - BucketTargetOption.predefinedAcl(PredefinedAcl.PROJECT_PRIVATE), - BucketTargetOption.predefinedDefaultObjectAcl(PredefinedAcl.PROJECT_PRIVATE)); - } - @Test public void testGetObjectIsRoutedToV2AndKeepsAllFields() throws IOException { GcsUtil gcsUtil = gcsUtilRoutingToV2(); @@ -2277,6 +2264,81 @@ public void testV2OnlyMethodsFailWithoutV2() { Mockito.verifyNoInteractions(v1); } + @Test + public void testExpandIsRoutedToV2() throws IOException { + GcsUtil gcsUtil = gcsUtilRoutingToV2(); + GcsPath pattern = GcsPath.fromUri("gs://bucket/prefix/*"); + List expanded = ImmutableList.of(GcsPath.fromUri("gs://bucket/prefix/a")); + when(mockDelegateV2.expand(pattern)).thenReturn(expanded); + + assertSame(expanded, gcsUtil.expand(pattern)); + Mockito.verifyNoMoreInteractions(mockDelegate); + } + + @Test + public void testFileSizeIsRoutedToV2() throws IOException { + GcsUtil gcsUtil = gcsUtilRoutingToV2(); + GcsPath path = GcsPath.fromUri("gs://bucket/object"); + when(mockDelegateV2.fileSize(path)).thenReturn(42L); + + assertEquals(42L, gcsUtil.fileSize(path)); + Mockito.verifyNoMoreInteractions(mockDelegate); + } + + /** + * Only the routing of the delimiter overload is checked. The page conversion is covered by {@link + * #testListObjectsIsRoutedToV2AndConvertsAPage}. + */ + @Test + public void testListObjectsWithDelimiterIsRoutedToV2() throws IOException { + GcsUtil gcsUtil = gcsUtilRoutingToV2(); + com.google.cloud.storage.Blob object = mockBlob("bucket", "prefix/object"); + @SuppressWarnings("unchecked") + Page page = Mockito.mock(Page.class); + when(page.getValues()).thenReturn(ImmutableList.of(object)); + when(mockDelegateV2.listBlobs("bucket", "prefix/", "token", "/")).thenReturn(page); + + Objects objects = gcsUtil.listObjects("bucket", "prefix/", "token", "/"); + + assertEquals("prefix/object", objects.getItems().get(0).getName()); + Mockito.verifyNoMoreInteractions(mockDelegate); + } + + @Test + public void testOpenIsRoutedToV2() throws IOException { + GcsUtil gcsUtil = gcsUtilRoutingToV2(); + GcsPath path = GcsPath.fromUri("gs://bucket/object"); + SeekableByteChannel channel = Mockito.mock(SeekableByteChannel.class); + when(mockDelegateV2.open(path)).thenReturn(channel); + + assertSame(channel, gcsUtil.open(path)); + Mockito.verifyNoMoreInteractions(mockDelegate); + } + + @Test + public void testVerifyBucketAccessibleIsRoutedToV2() throws IOException { + GcsUtil gcsUtil = gcsUtilRoutingToV2(); + GcsPath path = GcsPath.fromUri("gs://bucket/object"); + + gcsUtil.verifyBucketAccessible(path); + + verify(mockDelegateV2).verifyBucketAccessible(path); + Mockito.verifyNoMoreInteractions(mockDelegate); + } + + @Test + public void testBucketAccessibleIsRoutedToV2() throws IOException { + GcsUtil gcsUtil = gcsUtilRoutingToV2(); + GcsPath accessible = GcsPath.fromUri("gs://accessible/object"); + GcsPath inaccessible = GcsPath.fromUri("gs://inaccessible/object"); + when(mockDelegateV2.bucketAccessible(accessible)).thenReturn(true); + when(mockDelegateV2.bucketAccessible(inaccessible)).thenReturn(false); + + assertTrue(gcsUtil.bucketAccessible(accessible)); + assertFalse(gcsUtil.bucketAccessible(inaccessible)); + Mockito.verifyNoMoreInteractions(mockDelegate); + } + // The tests below exercise a real GcsUtilV2 delegate whose java-storage client is mocked, to // cover behavior that GcsUtilV2 must share with GcsUtilV1. // TODO: Move these to a parity test that runs against both delegates. @@ -2334,26 +2396,6 @@ public void testV2WriteChannelCloseTranslatesStorageException() throws IOExcepti assertSame(preconditionFailed, thrown.getCause()); } - /** Well-known failures on close are translated like any other V2 call, e.g. 403 and 404. */ - @Test - public void testV2WriteChannelCloseTranslatesKnownStatusCodes() throws IOException { - com.google.cloud.storage.Storage storage = Mockito.mock(com.google.cloud.storage.Storage.class); - WriteChannel writer = Mockito.mock(WriteChannel.class); - Mockito.doThrow(new StorageException(403, "Forbidden")) - .doThrow(new StorageException(404, "Not Found")) - .when(writer) - .close(); - when(storage.writer(any(com.google.cloud.storage.BlobInfo.class), any())).thenReturn(writer); - GcsUtil gcsUtil = gcsUtilWithV2Storage(storage); - GcsPath path = GcsPath.fromComponents("testbucket", "testobject"); - CreateOptions options = CreateOptions.builder().setExpectFileToNotExist(true).build(); - - WritableByteChannel forbidden = gcsUtil.create(path, options); - assertThrows(AccessDeniedException.class, forbidden::close); - WritableByteChannel notFound = gcsUtil.create(path, options); - assertThrows(FileNotFoundException.class, notFound::close); - } - /** Returns a {@link GcsUtil} backed by a real {@link GcsUtilV2} whose bucket lookup throws. */ private GcsUtil gcsUtilWithV2BucketLookupFailing(IOException failure) throws IOException { GcsOptions options = gcsOptionsWithTestCredential(); From 34d4009783bd4f4d1dd260830e9cfa237dfcef5d Mon Sep 17 00:00:00 2001 From: Shunping Huang Date: Fri, 25 Sep 2026 10:19:25 -0400 Subject: [PATCH 8/9] Make GcsUtilV2's storage client mockable in tests Route all client access through a @VisibleForTesting storage() accessor. In GcsUtilTest, merge the V2 helpers into one, test bucketAccessible through a mocked client, and reset metrics containers in tearDown. --- .../sdk/extensions/gcp/util/GcsUtilV2.java | 39 +++++++++------ .../sdk/extensions/gcp/util/GcsUtilTest.java | 50 +++++++++++-------- 2 files changed, 52 insertions(+), 37 deletions(-) diff --git a/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilV2.java b/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilV2.java index 15c3f728dacb..630f25e419e2 100644 --- a/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilV2.java +++ b/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilV2.java @@ -101,7 +101,16 @@ public GcsUtilV2 create(PipelineOptions options) { } } - private Storage storage; + private final Storage storage; + + /** + * The shared client. Every operation reaches it through this method, so that tests can substitute + * a mocked client for a spied instance. + */ + @VisibleForTesting + Storage storage() { + return storage; + } private final @Nullable Integer uploadBufferSizeBytes; @@ -303,13 +312,13 @@ public int hashCode() { @VisibleForTesting Storage storageWithHttpMetrics(@Nullable MetricsContainer container, boolean isWrite) { if (container == null) { - return storage; + return storage(); } - StorageOptions options = storage.getOptions(); + StorageOptions options = storage().getOptions(); TransportOptions transportOptions = options.getTransportOptions(); if (!(transportOptions instanceof HttpTransportOptions)) { // A non-HTTP transport (e.g. gRPC) has no HttpRequestInitializer to wrap. - return storage; + return storage(); } return options.toBuilder() .setTransportOptions( @@ -355,7 +364,7 @@ private IOException translateStorageException( } public Blob getBlob(GcsPath gcsPath, BlobGetOption... options) throws IOException { - return getBlob(storage, gcsPath, options); + return getBlob(storage(), gcsPath, options); } /** As {@link #getBlob(GcsPath, BlobGetOption...)}, but issued through a specific client. */ @@ -408,7 +417,7 @@ public List getBlobs(Iterable gcsPaths, BlobGetOption... op Lists.partition(Lists.newArrayList(gcsPaths), MAX_REQUESTS_PER_BATCH)) { // Create a new empty batch every time - StorageBatch batch = storage.batch(); + StorageBatch batch = storage().batch(); List> batchResultFutures = new ArrayList<>(); for (GcsPath path : pathPartition) { @@ -464,7 +473,7 @@ public Page listBlobs( } try { - return storage.list(bucket, blobListOptions.toArray(new BlobListOption[0])); + return storage().list(bucket, blobListOptions.toArray(new BlobListOption[0])); } catch (StorageException e) { throw translateStorageException(bucket, prefix, e); } @@ -534,7 +543,7 @@ public void remove(Iterable paths, MissingStrategy strategy) throws IOE Lists.partition(Lists.newArrayList(paths), MAX_REQUESTS_PER_BATCH)) { // Create a new empty batch every time - StorageBatch batch = storage.batch(); + StorageBatch batch = storage().batch(); List> batchResultFutures = new ArrayList<>(); for (GcsPath path : pathPartition) { @@ -601,7 +610,7 @@ private void rewriteHelper( // FAIL_IF_EXISTS, SKIP_IF_EXISTS and SAFE_OVERWRITE require checking the target blob BlobInfo existingTarget; try { - existingTarget = storage.get(dstId); + existingTarget = storage().get(dstId); } catch (StorageException e) { throw translateStorageException(dstPath, e); } @@ -631,11 +640,11 @@ private void rewriteHelper( } try { - CopyWriter copyWriter = storage.copy(copyRequestBuilder.build()); + CopyWriter copyWriter = storage().copy(copyRequestBuilder.build()); copyWriter.getResult(); if (deleteSrc) { - if (!storage.delete(srcId)) { + if (!storage().delete(srcId)) { // This may happen if the source file is deleted by another process after copy. LOG.warn( "Source file {} could not be deleted after move to {}. It may not have existed.", @@ -673,7 +682,7 @@ public void move( public Bucket getBucket(GcsPath path, BucketGetOption... options) throws IOException { String bucketName = path.getBucket(); try { - Bucket bucket = storage.get(bucketName, options); + Bucket bucket = storage().get(bucketName, options); if (bucket == null) { throw new FileNotFoundException( String.format("The specified bucket does not exist: gs://%s", bucketName)); @@ -730,12 +739,12 @@ public void createBucket(BucketInfo bucketInfo, BucketTargetOption... options) public void createBucket( @Nullable String projectId, BucketInfo bucketInfo, BucketTargetOption... options) throws IOException { - Storage client = storage; + Storage client = storage(); if (projectId != null && !projectId.equals(this.projectId)) { // The owning project is a property of the client rather than of the insert request, so // asking for a different one means deriving a client for it. The derived client shares the // credentials, host and transport of the original. - client = storage.getOptions().toBuilder().setProjectId(projectId).build().getService(); + client = storage().getOptions().toBuilder().setProjectId(projectId).build().getService(); } try { client.create(bucketInfo, options); @@ -746,7 +755,7 @@ public void createBucket( public void removeBucket(BucketInfo bucketInfo) throws IOException { try { - if (!storage.delete(bucketInfo.getName())) { + if (!storage().delete(bucketInfo.getName())) { throw new FileNotFoundException( String.format("The specified bucket does not exist: gs://%s", bucketInfo.getName())); } diff --git a/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilTest.java b/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilTest.java index eceaa15b7bd7..4beb56d6c200 100644 --- a/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilTest.java +++ b/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilTest.java @@ -30,7 +30,6 @@ import static org.junit.Assert.assertThrows; import static org.junit.Assert.assertTrue; import static org.mockito.ArgumentMatchers.any; -import static org.mockito.ArgumentMatchers.anyBoolean; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -71,6 +70,7 @@ import com.google.cloud.hadoop.gcsio.GoogleCloudStorageReadOptions; import com.google.cloud.hadoop.gcsio.StorageResourceId; import com.google.cloud.storage.BucketInfo; +import com.google.cloud.storage.Storage.BucketGetOption; import com.google.cloud.storage.Storage.BucketTargetOption; import com.google.cloud.storage.Storage.PredefinedAcl; import com.google.cloud.storage.StorageClass; @@ -122,6 +122,7 @@ import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Lists; import org.checkerframework.checker.nullness.qual.Nullable; +import org.junit.After; import org.junit.Before; import org.junit.Rule; import org.junit.Test; @@ -145,6 +146,13 @@ public void setUp() { MetricsEnvironment.setCurrentContainer(testMetricsContainer); } + @After + public void tearDown() { + // Don't leak the containers installed by setUp into later tests in the same JVM. + MetricsEnvironment.setProcessWideContainer(null); + MetricsEnvironment.setCurrentContainer(null); + } + private static GcsOptions gcsOptionsWithTestCredential() { GcsOptions pipelineOptions = PipelineOptionsFactory.as(GcsOptions.class); pipelineOptions.setGcpCredential(new TestCredential()); @@ -2343,13 +2351,17 @@ public void testBucketAccessibleIsRoutedToV2() throws IOException { // cover behavior that GcsUtilV2 must share with GcsUtilV1. // TODO: Move these to a parity test that runs against both delegates. - /** Returns a {@link GcsUtil} backed by a real {@link GcsUtilV2} that calls {@code storage}. */ + /** + * Returns a {@link GcsUtil} backed by a real {@link GcsUtilV2} that issues every call to {@code + * storage}. Performance metrics are off by default, so the per-operation clients of {@link + * GcsUtilV2#storageWithHttpMetrics} resolve to this one as well. + */ private GcsUtil gcsUtilWithV2Storage(com.google.cloud.storage.Storage storage) { GcsOptions options = gcsOptionsWithTestCredential(); options.setProject("my_project"); GcsUtil gcsUtil = options.getGcsUtil(); GcsUtilV2 delegateV2 = Mockito.spy(new GcsUtilV2(options)); - Mockito.doReturn(storage).when(delegateV2).storageWithHttpMetrics(any(), anyBoolean()); + Mockito.doReturn(storage).when(delegateV2).storage(); gcsUtil.delegateV2 = delegateV2; return gcsUtil; } @@ -2396,22 +2408,12 @@ public void testV2WriteChannelCloseTranslatesStorageException() throws IOExcepti assertSame(preconditionFailed, thrown.getCause()); } - /** Returns a {@link GcsUtil} backed by a real {@link GcsUtilV2} whose bucket lookup throws. */ - private GcsUtil gcsUtilWithV2BucketLookupFailing(IOException failure) throws IOException { - GcsOptions options = gcsOptionsWithTestCredential(); - options.setProject("my_project"); - GcsUtil gcsUtil = options.getGcsUtil(); - GcsUtilV2 delegateV2 = Mockito.spy(new GcsUtilV2(options)); - Mockito.doThrow(failure).when(delegateV2).getBucket(any(GcsPath.class), any()); - gcsUtil.delegateV2 = delegateV2; - return gcsUtil; - } - /** Mirrors {@link #testBucketDoesNotExist} for V2. */ @Test public void testV2BucketAccessibleIsFalseWhenBucketDoesNotExist() throws IOException { - GcsUtil gcsUtil = - gcsUtilWithV2BucketLookupFailing(new FileNotFoundException("gs://testbucket")); + com.google.cloud.storage.Storage storage = Mockito.mock(com.google.cloud.storage.Storage.class); + // An unstubbed get() returns null, which is how java-storage reports a missing bucket. + GcsUtil gcsUtil = gcsUtilWithV2Storage(storage); assertFalse(gcsUtil.bucketAccessible(GcsPath.fromComponents("testbucket", "testobject"))); } @@ -2419,8 +2421,10 @@ public void testV2BucketAccessibleIsFalseWhenBucketDoesNotExist() throws IOExcep /** Mirrors {@link #testBucketDoesNotExistBecauseOfAccessError} for V2. */ @Test public void testV2BucketAccessibleIsFalseWhenAccessIsDenied() throws IOException { - GcsUtil gcsUtil = - gcsUtilWithV2BucketLookupFailing(new AccessDeniedException("gs://testbucket")); + com.google.cloud.storage.Storage storage = Mockito.mock(com.google.cloud.storage.Storage.class); + when(storage.get(Mockito.eq("testbucket"), any(BucketGetOption.class))) + .thenThrow(new StorageException(403, "Forbidden")); + GcsUtil gcsUtil = gcsUtilWithV2Storage(storage); assertFalse(gcsUtil.bucketAccessible(GcsPath.fromComponents("testbucket", "testobject"))); } @@ -2430,14 +2434,16 @@ public void testV2BucketAccessibleIsFalseWhenAccessIsDenied() throws IOException * propagate rather than be reported as an inaccessible bucket, as V1 does. */ @Test - public void testV2BucketAccessiblePropagatesOtherFailures() throws IOException { - IOException serverError = new IOException(new StorageException(503, "Service Unavailable")); - GcsUtil gcsUtil = gcsUtilWithV2BucketLookupFailing(serverError); + public void testV2BucketAccessiblePropagatesOtherFailures() { + com.google.cloud.storage.Storage storage = Mockito.mock(com.google.cloud.storage.Storage.class); + StorageException serverError = new StorageException(503, "Service Unavailable"); + when(storage.get(Mockito.eq("testbucket"), any(BucketGetOption.class))).thenThrow(serverError); + GcsUtil gcsUtil = gcsUtilWithV2Storage(storage); IOException thrown = assertThrows( IOException.class, () -> gcsUtil.bucketAccessible(GcsPath.fromComponents("testbucket", "testobject"))); - assertSame(serverError, thrown); + assertSame(serverError, thrown.getCause()); } } From 5866744a1c2cd628f22176145505fade1f7604d3 Mon Sep 17 00:00:00 2001 From: Shunping Huang Date: Mon, 28 Sep 2026 10:02:55 -0400 Subject: [PATCH 9/9] Add reference link for default upload chunk size --- .../org/apache/beam/sdk/extensions/gcp/util/GcsUtilV2.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilV2.java b/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilV2.java index 630f25e419e2..20260aa3c79e 100644 --- a/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilV2.java +++ b/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilV2.java @@ -128,7 +128,8 @@ Storage storage() { /** * Upload chunk size applied when the pipeline does not ask for one. Mirrors gcsio's {@code - * AsyncWriteChannelOptions} default, which java-storage does not share. + * AsyncWriteChannelOptions} default, which java-storage does not share. Reference link: + * https://github.com/GoogleCloudDataproc/hadoop-connectors/blob/v3.1.14/util/src/main/java/com/google/cloud/hadoop/util/AsyncWriteChannelOptions.java#L72 */ @VisibleForTesting static final int DEFAULT_UPLOAD_CHUNK_SIZE_BYTES =