You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "shunping (via GitHub)" <gi...@apache.org> on 2026/02/04 16:45:41 UTC
[PR] GCS client library migration in Java SDK - part 2 [beam]
shunping opened a new pull request, #37502:
URL: https://github.com/apache/beam/pull/37502
(no comment)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GCS client library migration in Java SDK - part 2 [beam]
Posted by "shunping (via GitHub)" <gi...@apache.org>.
shunping commented on code in PR #37502:
URL: https://github.com/apache/beam/pull/37502#discussion_r2795100565
##########
sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilV2.java:
##########
@@ -0,0 +1,329 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.beam.sdk.extensions.gcp.util;
+
+import static org.apache.beam.sdk.io.FileSystemUtils.wildcardToRegexp;
+import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import com.google.api.gax.paging.Page;
+import com.google.auto.value.AutoValue;
+import com.google.cloud.storage.Blob;
+import com.google.cloud.storage.Bucket;
+import com.google.cloud.storage.BucketInfo;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.Storage.BlobField;
+import com.google.cloud.storage.Storage.BlobGetOption;
+import com.google.cloud.storage.Storage.BlobListOption;
+import com.google.cloud.storage.Storage.BucketField;
+import com.google.cloud.storage.Storage.BucketGetOption;
+import com.google.cloud.storage.StorageBatch;
+import com.google.cloud.storage.StorageBatchResult;
+import com.google.cloud.storage.StorageException;
+import com.google.cloud.storage.StorageOptions;
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.nio.file.AccessDeniedException;
+import java.nio.file.FileAlreadyExistsException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.regex.Pattern;
+import org.apache.beam.sdk.extensions.gcp.options.GcpOptions;
+import org.apache.beam.sdk.extensions.gcp.util.gcsfs.GcsPath;
+import org.apache.beam.sdk.options.DefaultValueFactory;
+import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
+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;
+
+class GcsUtilV2 {
+ private static final org.slf4j.Logger LOG = org.slf4j.LoggerFactory.getLogger(GcsUtilV2.class);
+
+ public static class GcsUtilFactory implements DefaultValueFactory<GcsUtilV2> {
+ @Override
+ public GcsUtilV2 create(PipelineOptions options) {
+ // GcsOptions gcsOptions = options.as(GcsOptions.class);
+ // Storage.Builder storageBuilder = Transport.newStorageClient(gcsOptions);
+ return new GcsUtilV2(options);
+ }
+ }
+
+ private Storage storage;
+
+ /** Maximum number of items to retrieve per Objects.List request. */
+ private static final long MAX_LIST_BLOBS_PER_CALL = 1024;
+
+ /** Maximum number of requests permitted in a GCS batch request. */
+ private static final int MAX_REQUESTS_PER_BATCH = 100;
+
+ GcsUtilV2(PipelineOptions options) {
+ String projectId = options.as(GcpOptions.class).getProject();
+ storage = StorageOptions.newBuilder().setProjectId(projectId).build().getService();
+ }
+
+ @SuppressWarnings({
+ "nullness" // For Creating AccessDeniedException and FileAlreadyExistsException with null.
+ })
+ private IOException translateStorageException(GcsPath gcsPath, StorageException e) {
+ switch (e.getCode()) {
+ case 403:
+ return new AccessDeniedException(gcsPath.toString(), null, e.getMessage());
+ case 409:
+ return new FileAlreadyExistsException(gcsPath.toString(), null, e.getMessage());
+ default:
+ return new IOException(e);
+ }
+ }
+
+ private IOException translateStorageException(
+ String bucketName, @Nullable String blobName, StorageException e) {
+ return translateStorageException(GcsPath.fromComponents(bucketName, blobName), e);
+ }
+
+ public Blob getBlob(GcsPath gcsPath, BlobGetOption... options) throws IOException {
+ try {
+ Blob blob = storage.get(gcsPath.getBucket(), gcsPath.getObject(), options);
+ if (blob == null) {
+ throw new FileNotFoundException(
+ String.format("The specified file does not exist: %s", gcsPath.toString()));
+ }
+ return blob;
+ } catch (StorageException e) {
+ throw translateStorageException(gcsPath, e);
+ }
+ }
+
+ public long fileSize(GcsPath gcsPath) throws IOException {
+ return getBlob(gcsPath, BlobGetOption.fields(BlobField.SIZE)).getSize();
+ }
+
+ /** A class that holds either a {@link Blob} or an {@link IOException}. */
+ // It is clear from the name that this class holds either Blob or IOException.
+ @SuppressFBWarnings("NM_CLASS_NOT_EXCEPTION")
+ @AutoValue
+ public abstract static class BlobOrIOException {
+
+ /** Returns the {@link Blob}. */
+ public abstract @Nullable Blob blob();
+
+ /** Returns the {@link IOException}. */
+ public abstract @Nullable IOException ioException();
+
+ @VisibleForTesting
+ public static BlobOrIOException create(Blob blob) {
+ return new AutoValue_GcsUtilV2_BlobOrIOException(
+ checkNotNull(blob, "blob"), null /* ioException */);
+ }
+
+ @VisibleForTesting
+ public static BlobOrIOException create(IOException ioException) {
+ return new AutoValue_GcsUtilV2_BlobOrIOException(
+ null /* blob */, checkNotNull(ioException, "ioException"));
+ }
+ }
+
+ public List<BlobOrIOException> getBlobs(Iterable<GcsPath> gcsPaths, BlobGetOption... options)
+ throws IOException {
+
+ List<BlobOrIOException> results = new ArrayList<>();
+
+ for (List<GcsPath> pathPartition :
+ Lists.partition(Lists.newArrayList(gcsPaths), MAX_REQUESTS_PER_BATCH)) {
+
+ // Create a new empty batch every time
+ StorageBatch batch = storage.batch();
+ List<StorageBatchResult<Blob>> batchResultFutures = new ArrayList<>();
+
+ for (GcsPath path : pathPartition) {
+ batchResultFutures.add(batch.get(path.getBucket(), path.getObject(), options));
+ }
+ batch.submit();
+
+ for (int i = 0; i < batchResultFutures.size(); i++) {
+ StorageBatchResult<Blob> future = batchResultFutures.get(i);
+ try {
+ Blob blob = future.get();
+ if (blob != null) {
+ results.add(BlobOrIOException.create(blob));
+ } else {
+ results.add(
+ BlobOrIOException.create(
+ new FileNotFoundException(
+ String.format(
+ "The specified file does not exist: %s",
+ pathPartition.get(i).toString()))));
+ }
+ } catch (StorageException e) {
+ results.add(BlobOrIOException.create(translateStorageException(pathPartition.get(i), e)));
+ }
+ }
+ }
+ return results;
+ }
+
+ /** Lists {@link Blob}s given the {@code bucket}, {@code prefix}, {@code pageToken}. */
+ public Page<Blob> listBlobs(
+ String bucket,
+ String prefix,
+ @Nullable String pageToken,
+ @Nullable String delimiter,
+ BlobListOption... options)
+ throws IOException {
+ List<BlobListOption> blobListOptions = new ArrayList<>();
+ blobListOptions.add(BlobListOption.pageSize(MAX_LIST_BLOBS_PER_CALL));
+ if (pageToken != null) {
+ blobListOptions.add(BlobListOption.pageToken(pageToken));
+ }
+ if (prefix != null) {
+ blobListOptions.add(BlobListOption.prefix(prefix));
+ }
+ if (delimiter != null) {
+ blobListOptions.add(BlobListOption.delimiter(delimiter));
+ }
+ if (options != null && options.length > 0) {
+ for (BlobListOption option : options) {
+ blobListOptions.add(option);
+ }
+ }
+
+ try {
+ return storage.list(bucket, blobListOptions.toArray(new BlobListOption[0]));
+ } catch (StorageException e) {
+ throw translateStorageException(bucket, prefix, e);
+ }
+ }
+
+ public Page<Blob> listBlobs(
+ String bucket, String prefix, @Nullable String pageToken, BlobListOption... options)
+ throws IOException {
+ return listBlobs(bucket, prefix, pageToken, null, options);
+ }
+
+ /**
+ * Expands a pattern into matched paths. The pattern path may contain globs, which are expanded in
+ * the result. For patterns that only match a single object, we ensure that the object exists.
+ */
+ public List<GcsPath> expand(GcsPath gcsPattern) throws IOException {
+ // Handle Non-Wildcard Path
+ if (!GcsPath.isWildcard(gcsPattern)) {
+ try {
+ // Use a get request to fetch the metadata of the object, and ignore the return value.
+ // The request has strong global consistency.
+ getBlob(gcsPattern, BlobGetOption.fields(BlobField.NAME));
+ return ImmutableList.of(gcsPattern);
+ } catch (FileNotFoundException e) {
+ // If the path was not found, return an empty list.
+ return ImmutableList.of();
+ }
+ }
+
+ // Handle Wildcard Path
+ // TODO: check out BlobListOption.matchGlob() for a similar function.
Review Comment:
I put it as a TODO, because this seems to be a beta feature.
<img width="1764" height="596" alt="image" src="https://github.com/user-attachments/assets/a25be397-0e1f-42bd-ad8e-b9747b3fe3cd" />
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GCS client library migration in Java SDK - part 2 [beam]
Posted by "gemini-code-assist[bot] (via GitHub)" <gi...@apache.org>.
gemini-code-assist[bot] commented on PR #37502:
URL: https://github.com/apache/beam/pull/37502#issuecomment-3848543345
## Summary of Changes
Hello @shunping, I'm Gemini Code Assist[^1]! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed!
This pull request is the second part of migrating the Java SDK's Google Cloud Storage client library. It introduces `GcsUtilV2`, a new utility class that integrates the modern `google-cloud-storage` client. The existing `GcsUtil` is enhanced to support both the legacy and new client implementations, with the choice controlled by an experimental flag. This approach enables a controlled transition to the updated GCS client, ensuring compatibility and allowing for thorough testing of the new implementation.
### Highlights
* **Introduction of GcsUtilV2**: A new class `GcsUtilV2` has been added to utilize the modern `google-cloud-storage` client library for GCS operations, providing updated methods for interacting with Google Cloud Storage.
* **Conditional GcsUtilV2 Usage**: The main `GcsUtil` class now conditionally initializes and delegates calls to `GcsUtilV2` based on an experimental flag (`use_gcsutil_v2`), allowing for a phased adoption and controlled rollout of the new client.
* **API Modernization**: Existing `getObject` methods in `GcsUtil` are deprecated in favor of a new `getBlob` method, aligning the API with the new client library's `Blob` object model for GCS objects.
* **Parameterized Integration Tests**: Integration tests (`GcsUtilIT`) have been refactored to run parameterized tests, verifying functionality against both the legacy (`GcsUtilV1`) and the new (`GcsUtilV2`) GCS utility implementations to ensure compatibility and correctness.
🧠**New Feature in Public Preview:** You can now enable **Memory** to help **Gemini Code Assist** learn from your team's feedback. This makes future code reviews more consistent and personalized to your project's style. **Click [here](https://codeassist.google/code-review/login) to enable Memory in your admin console.**
<details>
<summary><b>Changelog</b></summary>
* **sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtil.java**
* Added `com.google.cloud.storage.Blob` import.
* Introduced `delegateV2` field for `GcsUtilV2`.
* Modified constructor to accept `shouldUseV2` parameter, enabling conditional `GcsUtilV2` instantiation via an experimental flag.
* Updated `fileSize` to delegate to `delegateV2` if active.
* Deprecated `getObject` methods, recommending `getBlob(GcsPath)`.
* Added `getBlob(GcsPath)` method, delegating to `delegateV2` when enabled.
* **sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilV2.java**
* New file: Implements GCS operations using the `google-cloud-storage` client library.
* Provides `getBlob` and `fileSize` methods by interacting with `com.google.cloud.storage.Storage`.
* Initializes `Storage` client using `StorageOptions` and project ID from `PipelineOptions`.
* **sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilIT.java**
* Converted to a parameterized JUnit test (`@RunWith(Parameterized.class)`).
* Tests now run for both "use_gcsutil_v1" and "use_gcsutil_v2" experimental flags.
* `testFileSize` and `testGetObjectOrGetBlob` methods were updated to use the appropriate GCS utility version based on the active experiment.
* Several previous tests (`testRewriteMultiPart`, `testWriteAndReadGcsWithGrpc`) were commented out.
</details>
<details>
<summary><b>Activity</b></summary>
* No specific activity (comments, reviews, etc.) was provided in the context for this pull request.
</details>
<details>
<summary><b>Using Gemini Code Assist</b></summary>
<br>
The full guide for Gemini Code Assist can be found on our [documentation page](https://developers.google.com/gemini-code-assist/docs/review-github-code), here are some quick tips.
<b>Invoking Gemini</b>
You can request assistance from Gemini at any point by creating a comment using either `/gemini <command>` or `@gemini-code-assist <command>`. Below is a summary of the supported commands on the current page.
Feature | Command | Description
--- | --- | ---
Code Review | `/gemini review` | Performs a code review for the current pull request in its current state.
Pull Request Summary | `/gemini summary` | Provides a summary of the current pull request in its current state.
Comment | @gemini-code-assist | Responds in comments when explicitly tagged, both in pull request comments and review comments.
Help | `/gemini help` | Displays a list of available commands.
<b>Customization</b>
To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a `.gemini/` folder in the base of the repository. Detailed instructions can be found [here](https://developers.google.com/gemini-code-assist/docs/customize-gemini-behavior-github).
<b>Limitations & Feedback</b>
Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with :thumbsup: and :thumbsdown: on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for Github and other Google products, sign up [here](https://google.qualtrics.com/jfe/form/SV_2cyuGuTWsEw84yG).
<b>You can also get AI-powered code generation, chat, as well as code reviews directly in the IDE at no cost with the [Gemini Code Assist IDE Extension](https://cloud.google.com/products/gemini/code-assist).</b>
</details>
[^1]: Review the [Privacy Notices](https://policies.google.com/privacy), [Generative AI Prohibited Use Policy](https://policies.google.com/terms/generative-ai/use-policy), [Terms of Service](https://policies.google.com/terms), and learn how to configure Gemini Code Assist in GitHub [here](https://developers.google.com/gemini-code-assist/docs/customize-gemini-behavior-github). Gemini can make mistakes, so double check it and [use code with caution](https://support.google.com/legal/answer/13505487).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GCS client library migration in Java SDK - part 2 [beam]
Posted by "shunping (via GitHub)" <gi...@apache.org>.
shunping commented on PR #37502:
URL: https://github.com/apache/beam/pull/37502#issuecomment-3897613546
I verified that the new tests are running successfully under github workflow: https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java.yml
<img width="1472" height="836" alt="image" src="https://github.com/user-attachments/assets/44ac9a59-3db2-4d51-86d9-b76b20be3432" />
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GCS client library migration in Java SDK - part 2 [beam]
Posted by "shunping (via GitHub)" <gi...@apache.org>.
shunping commented on code in PR #37502:
URL: https://github.com/apache/beam/pull/37502#discussion_r2795089143
##########
sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtil.java:
##########
@@ -125,6 +134,36 @@ public static boolean isWildcard(GcsPath spec) {
rewriteDataOpBatchLimit,
gcsCountersOptions.delegate,
gcsOptions);
+ this.delegateV2 = null;
+ }
+
+ @VisibleForTesting
+ GcsUtil(
+ Storage storageClient,
+ HttpRequestInitializer httpRequestInitializer,
+ ExecutorService executorService,
+ Boolean shouldUseGrpc,
+ Credentials credentials,
+ @Nullable Integer uploadBufferSizeBytes,
+ @Nullable Integer rewriteDataOpBatchLimit,
+ GcsCountersOptions gcsCountersOptions,
+ GcsOptions gcsOptions,
+ Boolean shouldUseV2) {
Review Comment:
Sure, but notice that that this is a package-private constructor and it is not supposed to be called directly by users.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GCS client library migration in Java SDK - part 2 [beam]
Posted by "gemini-code-assist[bot] (via GitHub)" <gi...@apache.org>.
gemini-code-assist[bot] commented on code in PR #37502:
URL: https://github.com/apache/beam/pull/37502#discussion_r2784184733
##########
sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilV2.java:
##########
@@ -0,0 +1,330 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.beam.sdk.extensions.gcp.util;
+
+import static org.apache.beam.sdk.io.FileSystemUtils.wildcardToRegexp;
+import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import com.google.api.gax.paging.Page;
+import com.google.auto.value.AutoValue;
+import com.google.cloud.storage.Blob;
+import com.google.cloud.storage.Bucket;
+import com.google.cloud.storage.BucketInfo;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.Storage.BlobField;
+import com.google.cloud.storage.Storage.BlobGetOption;
+import com.google.cloud.storage.Storage.BlobListOption;
+import com.google.cloud.storage.Storage.BucketField;
+import com.google.cloud.storage.Storage.BucketGetOption;
+import com.google.cloud.storage.StorageBatch;
+import com.google.cloud.storage.StorageBatchResult;
+import com.google.cloud.storage.StorageException;
+import com.google.cloud.storage.StorageOptions;
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.nio.file.AccessDeniedException;
+import java.nio.file.FileAlreadyExistsException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.regex.Pattern;
+import org.apache.beam.sdk.extensions.gcp.options.GcpOptions;
+import org.apache.beam.sdk.extensions.gcp.util.gcsfs.GcsPath;
+import org.apache.beam.sdk.options.DefaultValueFactory;
+import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
+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;
+
+class GcsUtilV2 {
+ private static final org.slf4j.Logger LOG = org.slf4j.LoggerFactory.getLogger(GcsUtilV2.class);
+
+ public static class GcsUtilFactory implements DefaultValueFactory<GcsUtilV2> {
+ @Override
+ public GcsUtilV2 create(PipelineOptions options) {
+ // GcsOptions gcsOptions = options.as(GcsOptions.class);
+ // Storage.Builder storageBuilder = Transport.newStorageClient(gcsOptions);
+ return new GcsUtilV2(options);
+ }
+ }
+
+ private Storage storage;
+
+ /** Maximum number of items to retrieve per Objects.List request. */
+ private static final long MAX_LIST_BLOBS_PER_CALL = 1024;
+
+ /** Maximum number of requests permitted in a GCS batch request. */
+ private static final int MAX_REQUESTS_PER_BATCH = 100;
+
+ GcsUtilV2(PipelineOptions options) {
+ String projectId = options.as(GcpOptions.class).getProject();
+ storage = StorageOptions.newBuilder().setProjectId(projectId).build().getService();
+ }
+
+ @SuppressWarnings({
+ "nullness" // For Creating AccessDeniedException and FileAlreadyExistsException with null.
+ })
+ private IOException translateStorageException(GcsPath gcsPath, StorageException e) {
+ switch (e.getCode()) {
+ case 403:
+ return new AccessDeniedException(gcsPath.toString(), null, e.getMessage());
+ case 409:
+ return new FileAlreadyExistsException(gcsPath.toString(), null, e.getMessage());
+ default:
+ return new IOException(e);
+ }
+ }
+
+ private IOException translateStorageException(
+ String bucketName, @Nullable String blobName, StorageException e) {
+ return translateStorageException(GcsPath.fromComponents(bucketName, blobName), e);
+ }
+
+ public Blob getBlob(GcsPath gcsPath, BlobGetOption... options) throws IOException {
+ try {
+ Blob blob = storage.get(gcsPath.getBucket(), gcsPath.getObject(), options);
+ if (blob == null) {
+ throw new FileNotFoundException(
+ String.format("The specified file does not exist: %s", gcsPath.toString()));
+ }
+ return blob;
+ } catch (StorageException e) {
+ throw translateStorageException(gcsPath, e);
+ }
+ }
+
+ public long fileSize(GcsPath gcsPath) throws IOException {
+ return getBlob(gcsPath, BlobGetOption.fields(BlobField.SIZE)).getSize();
+ }
+
+ /** A class that holds either a {@link Blob} or an {@link IOException}. */
+ // It is clear from the name that this class holds either Blob or IOException.
+ @SuppressFBWarnings("NM_CLASS_NOT_EXCEPTION")
+ @AutoValue
+ public abstract static class BlobOrIOException {
+
+ /** Returns the {@link Blob}. */
+ public abstract @Nullable Blob blob();
+
+ /** Returns the {@link IOException}. */
+ public abstract @Nullable IOException ioException();
+
+ @VisibleForTesting
+ public static BlobOrIOException create(Blob blob) {
+ return new AutoValue_GcsUtilV2_BlobOrIOException(
+ checkNotNull(blob, "blob"), null /* ioException */);
+ }
+
+ @VisibleForTesting
+ public static BlobOrIOException create(IOException ioException) {
+ return new AutoValue_GcsUtilV2_BlobOrIOException(
+ null /* blob */, checkNotNull(ioException, "ioException"));
+ }
+ }
+
+ public List<BlobOrIOException> getBlobs(List<GcsPath> gcsPaths, BlobGetOption... options)
+ throws IOException {
+
+ List<BlobOrIOException> results = new ArrayList<>();
+
+ for (List<GcsPath> pathPartition :
+ Lists.partition(Lists.newArrayList(gcsPaths), MAX_REQUESTS_PER_BATCH)) {
+
+ // Create a new empty batch every time
+ StorageBatch batch = storage.batch();
+ List<StorageBatchResult<Blob>> batchResultFutures = new ArrayList<>();
+
+ for (GcsPath path : pathPartition) {
+ batchResultFutures.add(batch.get(path.getBucket(), path.getObject(), options));
+ }
+ batch.submit();
+
+ for (int i = 0; i < batchResultFutures.size(); i++) {
+ StorageBatchResult<Blob> future = batchResultFutures.get(i);
+ try {
+ Blob blob = future.get();
+ if (blob != null) {
+ results.add(BlobOrIOException.create(blob));
+ } else {
+ results.add(
+ BlobOrIOException.create(
+ new FileNotFoundException(
+ String.format(
+ "The specified file does not exist: %s", gcsPaths.get(i).toString()))));
+ }
+ } catch (StorageException e) {
+ results.add(BlobOrIOException.create(translateStorageException(gcsPaths.get(i), e)));
+ }
+ }
Review Comment:
There is a bug in how the `GcsPath` is retrieved for error reporting within the `getBlobs` method. The index `i` corresponds to the current `pathPartition`, not the original `gcsPaths` list. When processing multiple batches, this will result in incorrect file paths in `FileNotFoundException` messages and for `translateStorageException`. You should use `pathPartition.get(i)` instead of `gcsPaths.get(i)`.
```java
for (int i = 0; i < batchResultFutures.size(); i++) {
StorageBatchResult<Blob> future = batchResultFutures.get(i);
GcsPath gcsPath = pathPartition.get(i);
try {
Blob blob = future.get();
if (blob != null) {
results.add(BlobOrIOException.create(blob));
} else {
results.add(
BlobOrIOException.create(
new FileNotFoundException(
String.format(
"The specified file does not exist: %s", gcsPath.toString()))));
}
} catch (StorageException e) {
results.add(BlobOrIOException.create(translateStorageException(gcsPath, e)));
}
}
```
##########
sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilV2.java:
##########
@@ -0,0 +1,330 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.beam.sdk.extensions.gcp.util;
+
+import static org.apache.beam.sdk.io.FileSystemUtils.wildcardToRegexp;
+import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import com.google.api.gax.paging.Page;
+import com.google.auto.value.AutoValue;
+import com.google.cloud.storage.Blob;
+import com.google.cloud.storage.Bucket;
+import com.google.cloud.storage.BucketInfo;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.Storage.BlobField;
+import com.google.cloud.storage.Storage.BlobGetOption;
+import com.google.cloud.storage.Storage.BlobListOption;
+import com.google.cloud.storage.Storage.BucketField;
+import com.google.cloud.storage.Storage.BucketGetOption;
+import com.google.cloud.storage.StorageBatch;
+import com.google.cloud.storage.StorageBatchResult;
+import com.google.cloud.storage.StorageException;
+import com.google.cloud.storage.StorageOptions;
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.nio.file.AccessDeniedException;
+import java.nio.file.FileAlreadyExistsException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.regex.Pattern;
+import org.apache.beam.sdk.extensions.gcp.options.GcpOptions;
+import org.apache.beam.sdk.extensions.gcp.util.gcsfs.GcsPath;
+import org.apache.beam.sdk.options.DefaultValueFactory;
+import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
+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;
+
+class GcsUtilV2 {
+ private static final org.slf4j.Logger LOG = org.slf4j.LoggerFactory.getLogger(GcsUtilV2.class);
+
+ public static class GcsUtilFactory implements DefaultValueFactory<GcsUtilV2> {
+ @Override
+ public GcsUtilV2 create(PipelineOptions options) {
+ // GcsOptions gcsOptions = options.as(GcsOptions.class);
+ // Storage.Builder storageBuilder = Transport.newStorageClient(gcsOptions);
+ return new GcsUtilV2(options);
+ }
+ }
+
+ private Storage storage;
+
+ /** Maximum number of items to retrieve per Objects.List request. */
+ private static final long MAX_LIST_BLOBS_PER_CALL = 1024;
+
+ /** Maximum number of requests permitted in a GCS batch request. */
+ private static final int MAX_REQUESTS_PER_BATCH = 100;
+
+ GcsUtilV2(PipelineOptions options) {
+ String projectId = options.as(GcpOptions.class).getProject();
+ storage = StorageOptions.newBuilder().setProjectId(projectId).build().getService();
+ }
+
+ @SuppressWarnings({
+ "nullness" // For Creating AccessDeniedException and FileAlreadyExistsException with null.
+ })
+ private IOException translateStorageException(GcsPath gcsPath, StorageException e) {
+ switch (e.getCode()) {
+ case 403:
+ return new AccessDeniedException(gcsPath.toString(), null, e.getMessage());
+ case 409:
+ return new FileAlreadyExistsException(gcsPath.toString(), null, e.getMessage());
+ default:
+ return new IOException(e);
+ }
+ }
+
+ private IOException translateStorageException(
+ String bucketName, @Nullable String blobName, StorageException e) {
+ return translateStorageException(GcsPath.fromComponents(bucketName, blobName), e);
+ }
+
+ public Blob getBlob(GcsPath gcsPath, BlobGetOption... options) throws IOException {
+ try {
+ Blob blob = storage.get(gcsPath.getBucket(), gcsPath.getObject(), options);
+ if (blob == null) {
+ throw new FileNotFoundException(
+ String.format("The specified file does not exist: %s", gcsPath.toString()));
+ }
+ return blob;
+ } catch (StorageException e) {
+ throw translateStorageException(gcsPath, e);
+ }
+ }
+
+ public long fileSize(GcsPath gcsPath) throws IOException {
+ return getBlob(gcsPath, BlobGetOption.fields(BlobField.SIZE)).getSize();
+ }
+
+ /** A class that holds either a {@link Blob} or an {@link IOException}. */
+ // It is clear from the name that this class holds either Blob or IOException.
+ @SuppressFBWarnings("NM_CLASS_NOT_EXCEPTION")
+ @AutoValue
+ public abstract static class BlobOrIOException {
+
+ /** Returns the {@link Blob}. */
+ public abstract @Nullable Blob blob();
+
+ /** Returns the {@link IOException}. */
+ public abstract @Nullable IOException ioException();
+
+ @VisibleForTesting
+ public static BlobOrIOException create(Blob blob) {
+ return new AutoValue_GcsUtilV2_BlobOrIOException(
+ checkNotNull(blob, "blob"), null /* ioException */);
+ }
+
+ @VisibleForTesting
+ public static BlobOrIOException create(IOException ioException) {
+ return new AutoValue_GcsUtilV2_BlobOrIOException(
+ null /* blob */, checkNotNull(ioException, "ioException"));
+ }
+ }
+
+ public List<BlobOrIOException> getBlobs(List<GcsPath> gcsPaths, BlobGetOption... options)
+ throws IOException {
+
+ List<BlobOrIOException> results = new ArrayList<>();
+
+ for (List<GcsPath> pathPartition :
+ Lists.partition(Lists.newArrayList(gcsPaths), MAX_REQUESTS_PER_BATCH)) {
+
+ // Create a new empty batch every time
+ StorageBatch batch = storage.batch();
+ List<StorageBatchResult<Blob>> batchResultFutures = new ArrayList<>();
+
+ for (GcsPath path : pathPartition) {
+ batchResultFutures.add(batch.get(path.getBucket(), path.getObject(), options));
+ }
+ batch.submit();
+
+ for (int i = 0; i < batchResultFutures.size(); i++) {
+ StorageBatchResult<Blob> future = batchResultFutures.get(i);
+ try {
+ Blob blob = future.get();
+ if (blob != null) {
+ results.add(BlobOrIOException.create(blob));
+ } else {
+ results.add(
+ BlobOrIOException.create(
+ new FileNotFoundException(
+ String.format(
+ "The specified file does not exist: %s", gcsPaths.get(i).toString()))));
+ }
+ } catch (StorageException e) {
+ results.add(BlobOrIOException.create(translateStorageException(gcsPaths.get(i), e)));
+ }
+ }
+ }
+ return results;
+ }
+
+ /** Lists {@link Blob}s given the {@code bucket}, {@code prefix}, {@code pageToken}. */
+ public Page<Blob> listBlobs(
+ String bucket,
+ String prefix,
+ @Nullable String pageToken,
+ @Nullable String delimiter,
+ BlobListOption... options)
+ throws IOException {
+ List<BlobListOption> blobListOptions = new ArrayList<>();
+ blobListOptions.add(BlobListOption.pageSize(MAX_LIST_BLOBS_PER_CALL));
+ if (pageToken != null) {
+ blobListOptions.add(BlobListOption.pageToken(pageToken));
+ }
+ if (prefix != null) {
+ blobListOptions.add(BlobListOption.prefix(prefix));
+ }
+ if (delimiter != null) {
+ blobListOptions.add(BlobListOption.delimiter(delimiter));
+ }
+ if (options != null && options.length > 0) {
+ for (BlobListOption option : options) {
+ blobListOptions.add(option);
+ }
+ }
+
+ try {
+ return storage.list(bucket, blobListOptions.toArray(new BlobListOption[0]));
+ } catch (StorageException e) {
+ throw translateStorageException(bucket, prefix, e);
+ }
+ }
+
+ public Page<Blob> listBlobs(
+ String bucket, String prefix, @Nullable String pageToken, BlobListOption... options)
+ throws IOException {
+ return listBlobs(bucket, prefix, pageToken, null, options);
+ }
+
+ /**
+ * Expands a pattern into matched paths. The pattern path may contain globs, which are expanded in
+ * the result. For patterns that only match a single object, we ensure that the object exists.
+ */
+ public List<GcsPath> expand(GcsPath gcsPattern) throws IOException {
+ // Handle Non-Wildcard Path
+ if (!GcsPath.isWildcard(gcsPattern)) {
+ try {
+ // Use a get request to fetch the metadata of the object, and ignore the return value.
+ // The request has strong global consistency.
+ getBlob(gcsPattern, BlobGetOption.fields(BlobField.NAME));
+ return ImmutableList.of(gcsPattern);
+ } catch (FileNotFoundException e) {
+ // If the path was not found, return an empty list.
+ return ImmutableList.of();
+ }
+ }
+
+ // Handle Wildcard Path
+ // TODO: check out BlobListOption.matchGlob() for a similar function.
+ String prefix = GcsPath.getNonWildcardPrefix(gcsPattern.getObject());
+ Pattern p = Pattern.compile(wildcardToRegexp(gcsPattern.getObject()));
+
+ LOG.debug(
+ "matching files in bucket {}, prefix {} against pattern {}",
+ gcsPattern.getBucket(),
+ prefix,
+ p.toString());
+
+ List<GcsPath> results = new ArrayList<>();
+ Page<Blob> blobs =
+ listBlobs(
+ gcsPattern.getBucket(),
+ prefix,
+ null,
+ BlobListOption.fields(BlobField.NAME, BlobField.BUCKET));
+ // Iterate through all elements page by page (lazily)
+ for (Blob b : blobs.iterateAll()) {
+ String name = b.getName();
+ // Filter objects based on the regex. Skip directories, which end with a slash.
+ if (p.matcher(name).matches() && !name.endsWith("/")) {
+ LOG.debug("Matched object: {}", name);
+ results.add(GcsPath.fromComponents(b.getBucket(), b.getName()));
+ }
+ }
+ return results;
+ }
+
+ /** Get the {@link Bucket} from Cloud Storage path or propagates an exception. */
+ public Bucket getBucket(GcsPath path, BucketGetOption... options) throws IOException {
+ String bucketName = path.getBucket();
+ try {
+ Bucket bucket = storage.get(bucketName, options);
+ if (bucket == null) {
+ throw new FileNotFoundException(
+ String.format("The specified bucket does not exist: gs://%s", bucketName));
+ }
+ return bucket;
+ } catch (StorageException e) {
+ throw translateStorageException(bucketName, null, e);
+ }
+ }
+
+ /** Returns whether the GCS bucket exists and is accessible. */
+ public boolean bucketAccessible(GcsPath path) {
+ try {
+ // Fetch only the name field to minimize data transfer
+ getBucket(path, BucketGetOption.fields(BucketField.NAME));
+ return true;
+ } catch (IOException e) {
+ return false;
+ }
+ }
+
+ /**
+ * Checks whether the GCS bucket exists. Similar to {@link #bucketAccessible(GcsPath)}, but throws
+ * exception if the bucket is inaccessible due to permissions or does not exist.
+ */
+ public void verifyBucketAccessible(GcsPath path) throws IOException {
+ // Fetch only the name field to minimize data transfer
+ getBucket(path, BucketGetOption.fields(BucketField.NAME));
+ }
+
+ /**
+ * Returns the project number of the project which owns this bucket. If the bucket exists, it must
+ * be accessible otherwise the permissions exception will be propagated. If the bucket does not
+ * exist, an exception will be thrown.
+ */
+ public long bucketProject(GcsPath path) throws IOException {
+ Bucket bucket = getBucket(path, BucketGetOption.fields(BucketField.PROJECT));
+ return bucket.getProject().longValue();
+ }
+
+ public void createBucket(BucketInfo bucketInfo) throws IOException {
+ try {
+ storage.create(bucketInfo);
+ } catch (StorageException e) {
+ throw translateStorageException(bucketInfo.getName(), null, e);
+ }
+ }
+
+ public void removeBucket(BucketInfo bucketInfo) throws IOException {
+ Bucket bucket =
+ getBucket(
+ GcsPath.fromComponents(bucketInfo.getName(), null),
+ BucketGetOption.fields(BucketField.NAME));
+
+ try {
+ bucket.delete();
+ } catch (StorageException e) {
+ throw translateStorageException(bucketInfo.getName(), null, e);
+ }
+ }
Review Comment:
The current implementation of `removeBucket` first performs a `get` request to check if the bucket exists, and then a `delete` request. This results in two network calls and has a potential race condition if the bucket is deleted between the two calls.
This can be simplified to a single `storage.delete()` call which returns `false` if the bucket does not exist. This is more efficient and avoids the race condition.
```java
public void removeBucket(BucketInfo bucketInfo) throws IOException {
try {
if (!storage.delete(bucketInfo.getName())) {
throw new FileNotFoundException(
String.format("The specified bucket does not exist: gs://%s", bucketInfo.getName()));
}
} catch (StorageException e) {
throw translateStorageException(bucketInfo.getName(), null, e);
}
}
```
##########
sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtil.java:
##########
@@ -146,36 +188,86 @@ Integer getUploadBufferSizeBytes() {
}
public long fileSize(GcsPath path) throws IOException {
+ if (delegateV2 != null) {
+ return delegateV2.fileSize(path);
+ }
return delegate.fileSize(path);
}
+ /** @deprecated use {@link #getBlob(GcsPath)}. */
+ @Deprecated
public StorageObject getObject(GcsPath gcsPath) throws IOException {
return delegate.getObject(gcsPath);
}
+ /** @deprecated use {@link #getBlob(GcsPath, BlobGetOption...)}. */
+ @Deprecated
@VisibleForTesting
StorageObject getObject(GcsPath gcsPath, BackOff backoff, Sleeper sleeper) throws IOException {
return delegate.getObject(gcsPath, backoff, sleeper);
}
+ public Blob getBlob(GcsPath gcsPath, BlobGetOption... options) throws IOException {
+ if (delegateV2 != null) {
+ return delegateV2.getBlob(gcsPath, options);
+ }
+ throw new IOException("GcsUtil2 not initialized.");
Review Comment:
The error message "GcsUtil2 not initialized." contains a typo and should be "GcsUtilV2 not initialized." to be consistent with the class name `GcsUtilV2`. This typo is repeated in several other new methods that delegate to `GcsUtilV2` (`getBlobs`, `listBlobs`, `createBucket`, `getBucketWithOptions`, `removeBucket`).
```suggestion
throw new IOException("GcsUtilV2 not initialized.");
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GCS client library migration in Java SDK - part 2 [beam]
Posted by "shunping (via GitHub)" <gi...@apache.org>.
shunping commented on code in PR #37502:
URL: https://github.com/apache/beam/pull/37502#discussion_r2795135522
##########
sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/gcsfs/GcsPathTest.java:
##########
@@ -348,4 +348,24 @@ public void testSubPathError() {
a.subpath(1, 1); // throws IllegalArgumentException
Assert.fail();
}
+
+ @Test
+ public void testIsWildcard() {
+ assertTrue(GcsPath.isWildcard(GcsPath.fromUri("gs://bucket/foo*")));
+ assertTrue(GcsPath.isWildcard(GcsPath.fromUri("gs://bucket/foo?")));
+ assertTrue(GcsPath.isWildcard(GcsPath.fromUri("gs://bucket/foo[a-z]")));
Review Comment:
No behavior change here.
`isWildcard()` is only moved from GcsUtil into GcsPath, so it can be shared by GcsUtil V1 and V2.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GCS client library migration in Java SDK - part 2 [beam]
Posted by "Abacn (via GitHub)" <gi...@apache.org>.
Abacn commented on PR #37502:
URL: https://github.com/apache/beam/pull/37502#issuecomment-3873461394
/gemini review
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GCS client library migration in Java SDK - part 2 [beam]
Posted by "Abacn (via GitHub)" <gi...@apache.org>.
Abacn commented on code in PR #37502:
URL: https://github.com/apache/beam/pull/37502#discussion_r2794462072
##########
sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtil.java:
##########
@@ -125,6 +134,36 @@ public static boolean isWildcard(GcsPath spec) {
rewriteDataOpBatchLimit,
gcsCountersOptions.delegate,
gcsOptions);
+ this.delegateV2 = null;
+ }
+
+ @VisibleForTesting
+ GcsUtil(
+ Storage storageClient,
+ HttpRequestInitializer httpRequestInitializer,
+ ExecutorService executorService,
+ Boolean shouldUseGrpc,
+ Credentials credentials,
+ @Nullable Integer uploadBufferSizeBytes,
+ @Nullable Integer rewriteDataOpBatchLimit,
+ GcsCountersOptions gcsCountersOptions,
+ GcsOptions gcsOptions,
+ Boolean shouldUseV2) {
Review Comment:
we can avoid changing constructor signature of `GcsUtil` (then changing back) by moving `ExperimentalOptions.hasExperiment(options, "use_gcsutil_v2")` here inside constructor
##########
sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilV2.java:
##########
@@ -0,0 +1,329 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.beam.sdk.extensions.gcp.util;
+
+import static org.apache.beam.sdk.io.FileSystemUtils.wildcardToRegexp;
+import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import com.google.api.gax.paging.Page;
+import com.google.auto.value.AutoValue;
+import com.google.cloud.storage.Blob;
+import com.google.cloud.storage.Bucket;
+import com.google.cloud.storage.BucketInfo;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.Storage.BlobField;
+import com.google.cloud.storage.Storage.BlobGetOption;
+import com.google.cloud.storage.Storage.BlobListOption;
+import com.google.cloud.storage.Storage.BucketField;
+import com.google.cloud.storage.Storage.BucketGetOption;
+import com.google.cloud.storage.StorageBatch;
+import com.google.cloud.storage.StorageBatchResult;
+import com.google.cloud.storage.StorageException;
+import com.google.cloud.storage.StorageOptions;
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.nio.file.AccessDeniedException;
+import java.nio.file.FileAlreadyExistsException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.regex.Pattern;
+import org.apache.beam.sdk.extensions.gcp.options.GcpOptions;
+import org.apache.beam.sdk.extensions.gcp.util.gcsfs.GcsPath;
+import org.apache.beam.sdk.options.DefaultValueFactory;
+import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
+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;
+
+class GcsUtilV2 {
+ private static final org.slf4j.Logger LOG = org.slf4j.LoggerFactory.getLogger(GcsUtilV2.class);
+
+ public static class GcsUtilFactory implements DefaultValueFactory<GcsUtilV2> {
+ @Override
+ public GcsUtilV2 create(PipelineOptions options) {
+ // GcsOptions gcsOptions = options.as(GcsOptions.class);
+ // Storage.Builder storageBuilder = Transport.newStorageClient(gcsOptions);
+ return new GcsUtilV2(options);
+ }
+ }
+
+ private Storage storage;
+
+ /** Maximum number of items to retrieve per Objects.List request. */
+ private static final long MAX_LIST_BLOBS_PER_CALL = 1024;
+
+ /** Maximum number of requests permitted in a GCS batch request. */
+ private static final int MAX_REQUESTS_PER_BATCH = 100;
+
+ GcsUtilV2(PipelineOptions options) {
+ String projectId = options.as(GcpOptions.class).getProject();
+ storage = StorageOptions.newBuilder().setProjectId(projectId).build().getService();
+ }
+
+ @SuppressWarnings({
+ "nullness" // For Creating AccessDeniedException and FileAlreadyExistsException with null.
+ })
+ private IOException translateStorageException(GcsPath gcsPath, StorageException e) {
+ switch (e.getCode()) {
+ case 403:
+ return new AccessDeniedException(gcsPath.toString(), null, e.getMessage());
+ case 409:
+ return new FileAlreadyExistsException(gcsPath.toString(), null, e.getMessage());
+ default:
+ return new IOException(e);
+ }
+ }
+
+ private IOException translateStorageException(
+ String bucketName, @Nullable String blobName, StorageException e) {
+ return translateStorageException(GcsPath.fromComponents(bucketName, blobName), e);
+ }
+
+ public Blob getBlob(GcsPath gcsPath, BlobGetOption... options) throws IOException {
+ try {
+ Blob blob = storage.get(gcsPath.getBucket(), gcsPath.getObject(), options);
+ if (blob == null) {
+ throw new FileNotFoundException(
+ String.format("The specified file does not exist: %s", gcsPath.toString()));
+ }
+ return blob;
+ } catch (StorageException e) {
+ throw translateStorageException(gcsPath, e);
+ }
+ }
+
+ public long fileSize(GcsPath gcsPath) throws IOException {
+ return getBlob(gcsPath, BlobGetOption.fields(BlobField.SIZE)).getSize();
+ }
+
+ /** A class that holds either a {@link Blob} or an {@link IOException}. */
+ // It is clear from the name that this class holds either Blob or IOException.
+ @SuppressFBWarnings("NM_CLASS_NOT_EXCEPTION")
Review Comment:
Let's respect bugspot and rename the class, e.g., "BlobResult", which follows Java convention.
##########
sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilV2.java:
##########
@@ -0,0 +1,329 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.beam.sdk.extensions.gcp.util;
+
+import static org.apache.beam.sdk.io.FileSystemUtils.wildcardToRegexp;
+import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import com.google.api.gax.paging.Page;
+import com.google.auto.value.AutoValue;
+import com.google.cloud.storage.Blob;
+import com.google.cloud.storage.Bucket;
+import com.google.cloud.storage.BucketInfo;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.Storage.BlobField;
+import com.google.cloud.storage.Storage.BlobGetOption;
+import com.google.cloud.storage.Storage.BlobListOption;
+import com.google.cloud.storage.Storage.BucketField;
+import com.google.cloud.storage.Storage.BucketGetOption;
+import com.google.cloud.storage.StorageBatch;
+import com.google.cloud.storage.StorageBatchResult;
+import com.google.cloud.storage.StorageException;
+import com.google.cloud.storage.StorageOptions;
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.nio.file.AccessDeniedException;
+import java.nio.file.FileAlreadyExistsException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.regex.Pattern;
+import org.apache.beam.sdk.extensions.gcp.options.GcpOptions;
+import org.apache.beam.sdk.extensions.gcp.util.gcsfs.GcsPath;
+import org.apache.beam.sdk.options.DefaultValueFactory;
+import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
+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;
+
+class GcsUtilV2 {
+ private static final org.slf4j.Logger LOG = org.slf4j.LoggerFactory.getLogger(GcsUtilV2.class);
+
+ public static class GcsUtilFactory implements DefaultValueFactory<GcsUtilV2> {
+ @Override
+ public GcsUtilV2 create(PipelineOptions options) {
+ // GcsOptions gcsOptions = options.as(GcsOptions.class);
+ // Storage.Builder storageBuilder = Transport.newStorageClient(gcsOptions);
+ return new GcsUtilV2(options);
+ }
+ }
+
+ private Storage storage;
+
+ /** Maximum number of items to retrieve per Objects.List request. */
+ private static final long MAX_LIST_BLOBS_PER_CALL = 1024;
+
+ /** Maximum number of requests permitted in a GCS batch request. */
+ private static final int MAX_REQUESTS_PER_BATCH = 100;
+
+ GcsUtilV2(PipelineOptions options) {
+ String projectId = options.as(GcpOptions.class).getProject();
+ storage = StorageOptions.newBuilder().setProjectId(projectId).build().getService();
+ }
+
+ @SuppressWarnings({
+ "nullness" // For Creating AccessDeniedException and FileAlreadyExistsException with null.
+ })
+ private IOException translateStorageException(GcsPath gcsPath, StorageException e) {
+ switch (e.getCode()) {
+ case 403:
+ return new AccessDeniedException(gcsPath.toString(), null, e.getMessage());
+ case 409:
+ return new FileAlreadyExistsException(gcsPath.toString(), null, e.getMessage());
+ default:
+ return new IOException(e);
+ }
+ }
+
+ private IOException translateStorageException(
+ String bucketName, @Nullable String blobName, StorageException e) {
+ return translateStorageException(GcsPath.fromComponents(bucketName, blobName), e);
+ }
+
+ public Blob getBlob(GcsPath gcsPath, BlobGetOption... options) throws IOException {
+ try {
+ Blob blob = storage.get(gcsPath.getBucket(), gcsPath.getObject(), options);
+ if (blob == null) {
+ throw new FileNotFoundException(
+ String.format("The specified file does not exist: %s", gcsPath.toString()));
+ }
+ return blob;
+ } catch (StorageException e) {
+ throw translateStorageException(gcsPath, e);
+ }
+ }
+
+ public long fileSize(GcsPath gcsPath) throws IOException {
+ return getBlob(gcsPath, BlobGetOption.fields(BlobField.SIZE)).getSize();
+ }
+
+ /** A class that holds either a {@link Blob} or an {@link IOException}. */
+ // It is clear from the name that this class holds either Blob or IOException.
+ @SuppressFBWarnings("NM_CLASS_NOT_EXCEPTION")
+ @AutoValue
+ public abstract static class BlobOrIOException {
+
+ /** Returns the {@link Blob}. */
+ public abstract @Nullable Blob blob();
+
+ /** Returns the {@link IOException}. */
+ public abstract @Nullable IOException ioException();
+
+ @VisibleForTesting
+ public static BlobOrIOException create(Blob blob) {
+ return new AutoValue_GcsUtilV2_BlobOrIOException(
+ checkNotNull(blob, "blob"), null /* ioException */);
+ }
+
+ @VisibleForTesting
+ public static BlobOrIOException create(IOException ioException) {
+ return new AutoValue_GcsUtilV2_BlobOrIOException(
+ null /* blob */, checkNotNull(ioException, "ioException"));
+ }
+ }
+
+ public List<BlobOrIOException> getBlobs(Iterable<GcsPath> gcsPaths, BlobGetOption... options)
+ throws IOException {
+
+ List<BlobOrIOException> results = new ArrayList<>();
+
+ for (List<GcsPath> pathPartition :
+ Lists.partition(Lists.newArrayList(gcsPaths), MAX_REQUESTS_PER_BATCH)) {
+
+ // Create a new empty batch every time
+ StorageBatch batch = storage.batch();
+ List<StorageBatchResult<Blob>> batchResultFutures = new ArrayList<>();
+
+ for (GcsPath path : pathPartition) {
+ batchResultFutures.add(batch.get(path.getBucket(), path.getObject(), options));
+ }
+ batch.submit();
+
+ for (int i = 0; i < batchResultFutures.size(); i++) {
+ StorageBatchResult<Blob> future = batchResultFutures.get(i);
+ try {
+ Blob blob = future.get();
+ if (blob != null) {
+ results.add(BlobOrIOException.create(blob));
+ } else {
+ results.add(
+ BlobOrIOException.create(
+ new FileNotFoundException(
+ String.format(
+ "The specified file does not exist: %s",
+ pathPartition.get(i).toString()))));
+ }
+ } catch (StorageException e) {
+ results.add(BlobOrIOException.create(translateStorageException(pathPartition.get(i), e)));
+ }
+ }
+ }
+ return results;
+ }
+
+ /** Lists {@link Blob}s given the {@code bucket}, {@code prefix}, {@code pageToken}. */
+ public Page<Blob> listBlobs(
+ String bucket,
+ String prefix,
+ @Nullable String pageToken,
+ @Nullable String delimiter,
+ BlobListOption... options)
+ throws IOException {
+ List<BlobListOption> blobListOptions = new ArrayList<>();
+ blobListOptions.add(BlobListOption.pageSize(MAX_LIST_BLOBS_PER_CALL));
+ if (pageToken != null) {
+ blobListOptions.add(BlobListOption.pageToken(pageToken));
+ }
+ if (prefix != null) {
+ blobListOptions.add(BlobListOption.prefix(prefix));
+ }
+ if (delimiter != null) {
+ blobListOptions.add(BlobListOption.delimiter(delimiter));
+ }
+ if (options != null && options.length > 0) {
+ for (BlobListOption option : options) {
+ blobListOptions.add(option);
+ }
+ }
+
+ try {
+ return storage.list(bucket, blobListOptions.toArray(new BlobListOption[0]));
+ } catch (StorageException e) {
+ throw translateStorageException(bucket, prefix, e);
+ }
+ }
+
+ public Page<Blob> listBlobs(
+ String bucket, String prefix, @Nullable String pageToken, BlobListOption... options)
+ throws IOException {
+ return listBlobs(bucket, prefix, pageToken, null, options);
+ }
+
+ /**
+ * Expands a pattern into matched paths. The pattern path may contain globs, which are expanded in
+ * the result. For patterns that only match a single object, we ensure that the object exists.
+ */
+ public List<GcsPath> expand(GcsPath gcsPattern) throws IOException {
+ // Handle Non-Wildcard Path
+ if (!GcsPath.isWildcard(gcsPattern)) {
+ try {
+ // Use a get request to fetch the metadata of the object, and ignore the return value.
+ // The request has strong global consistency.
+ getBlob(gcsPattern, BlobGetOption.fields(BlobField.NAME));
+ return ImmutableList.of(gcsPattern);
+ } catch (FileNotFoundException e) {
+ // If the path was not found, return an empty list.
+ return ImmutableList.of();
+ }
+ }
+
+ // Handle Wildcard Path
+ // TODO: check out BlobListOption.matchGlob() for a similar function.
Review Comment:
Sounds like this TODO can be answered?
##########
sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/gcsfs/GcsPathTest.java:
##########
@@ -348,4 +348,24 @@ public void testSubPathError() {
a.subpath(1, 1); // throws IllegalArgumentException
Assert.fail();
}
+
+ @Test
+ public void testIsWildcard() {
+ assertTrue(GcsPath.isWildcard(GcsPath.fromUri("gs://bucket/foo*")));
+ assertTrue(GcsPath.isWildcard(GcsPath.fromUri("gs://bucket/foo?")));
+ assertTrue(GcsPath.isWildcard(GcsPath.fromUri("gs://bucket/foo[a-z]")));
Review Comment:
Interesting test case. Do we support regex wildcard "[a-z]" for GcsUtil V1? Or this is a result of implementation of GcsUtilV2
Wondering if previously `gs://bucket/foo[a-z]` was interpret as a normal file name, then this is a behavior change.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GCS client library migration in Java SDK - part 2 [beam]
Posted by "shunping (via GitHub)" <gi...@apache.org>.
shunping commented on code in PR #37502:
URL: https://github.com/apache/beam/pull/37502#discussion_r2795400521
##########
sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtil.java:
##########
@@ -125,6 +134,36 @@ public static boolean isWildcard(GcsPath spec) {
rewriteDataOpBatchLimit,
gcsCountersOptions.delegate,
gcsOptions);
+ this.delegateV2 = null;
+ }
+
+ @VisibleForTesting
+ GcsUtil(
+ Storage storageClient,
+ HttpRequestInitializer httpRequestInitializer,
+ ExecutorService executorService,
+ Boolean shouldUseGrpc,
+ Credentials credentials,
+ @Nullable Integer uploadBufferSizeBytes,
+ @Nullable Integer rewriteDataOpBatchLimit,
+ GcsCountersOptions gcsCountersOptions,
+ GcsOptions gcsOptions,
+ Boolean shouldUseV2) {
Review Comment:
Fixed.
##########
sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilV2.java:
##########
@@ -0,0 +1,329 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.beam.sdk.extensions.gcp.util;
+
+import static org.apache.beam.sdk.io.FileSystemUtils.wildcardToRegexp;
+import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import com.google.api.gax.paging.Page;
+import com.google.auto.value.AutoValue;
+import com.google.cloud.storage.Blob;
+import com.google.cloud.storage.Bucket;
+import com.google.cloud.storage.BucketInfo;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.Storage.BlobField;
+import com.google.cloud.storage.Storage.BlobGetOption;
+import com.google.cloud.storage.Storage.BlobListOption;
+import com.google.cloud.storage.Storage.BucketField;
+import com.google.cloud.storage.Storage.BucketGetOption;
+import com.google.cloud.storage.StorageBatch;
+import com.google.cloud.storage.StorageBatchResult;
+import com.google.cloud.storage.StorageException;
+import com.google.cloud.storage.StorageOptions;
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.nio.file.AccessDeniedException;
+import java.nio.file.FileAlreadyExistsException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.regex.Pattern;
+import org.apache.beam.sdk.extensions.gcp.options.GcpOptions;
+import org.apache.beam.sdk.extensions.gcp.util.gcsfs.GcsPath;
+import org.apache.beam.sdk.options.DefaultValueFactory;
+import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
+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;
+
+class GcsUtilV2 {
+ private static final org.slf4j.Logger LOG = org.slf4j.LoggerFactory.getLogger(GcsUtilV2.class);
+
+ public static class GcsUtilFactory implements DefaultValueFactory<GcsUtilV2> {
+ @Override
+ public GcsUtilV2 create(PipelineOptions options) {
+ // GcsOptions gcsOptions = options.as(GcsOptions.class);
+ // Storage.Builder storageBuilder = Transport.newStorageClient(gcsOptions);
+ return new GcsUtilV2(options);
+ }
+ }
+
+ private Storage storage;
+
+ /** Maximum number of items to retrieve per Objects.List request. */
+ private static final long MAX_LIST_BLOBS_PER_CALL = 1024;
+
+ /** Maximum number of requests permitted in a GCS batch request. */
+ private static final int MAX_REQUESTS_PER_BATCH = 100;
+
+ GcsUtilV2(PipelineOptions options) {
+ String projectId = options.as(GcpOptions.class).getProject();
+ storage = StorageOptions.newBuilder().setProjectId(projectId).build().getService();
+ }
+
+ @SuppressWarnings({
+ "nullness" // For Creating AccessDeniedException and FileAlreadyExistsException with null.
+ })
+ private IOException translateStorageException(GcsPath gcsPath, StorageException e) {
+ switch (e.getCode()) {
+ case 403:
+ return new AccessDeniedException(gcsPath.toString(), null, e.getMessage());
+ case 409:
+ return new FileAlreadyExistsException(gcsPath.toString(), null, e.getMessage());
+ default:
+ return new IOException(e);
+ }
+ }
+
+ private IOException translateStorageException(
+ String bucketName, @Nullable String blobName, StorageException e) {
+ return translateStorageException(GcsPath.fromComponents(bucketName, blobName), e);
+ }
+
+ public Blob getBlob(GcsPath gcsPath, BlobGetOption... options) throws IOException {
+ try {
+ Blob blob = storage.get(gcsPath.getBucket(), gcsPath.getObject(), options);
+ if (blob == null) {
+ throw new FileNotFoundException(
+ String.format("The specified file does not exist: %s", gcsPath.toString()));
+ }
+ return blob;
+ } catch (StorageException e) {
+ throw translateStorageException(gcsPath, e);
+ }
+ }
+
+ public long fileSize(GcsPath gcsPath) throws IOException {
+ return getBlob(gcsPath, BlobGetOption.fields(BlobField.SIZE)).getSize();
+ }
+
+ /** A class that holds either a {@link Blob} or an {@link IOException}. */
+ // It is clear from the name that this class holds either Blob or IOException.
+ @SuppressFBWarnings("NM_CLASS_NOT_EXCEPTION")
Review Comment:
Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GCS client library migration in Java SDK - part 2 [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #37502:
URL: https://github.com/apache/beam/pull/37502#issuecomment-3873471700
Assigning reviewers:
R: @Abacn for label java.
R: @Abacn for label build.
Note: If you would like to opt out of this review, comment `assign to next reviewer`.
Available commands:
- `stop reviewer notifications` - opt out of the automated review tooling
- `remind me after tests pass` - tag the comment author after tests pass
- `waiting on author` - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)
The PR bot will only process comments in the main thread (not review comments).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GCS client library migration in Java SDK - part 2 [beam]
Posted by "shunping (via GitHub)" <gi...@apache.org>.
shunping commented on code in PR #37502:
URL: https://github.com/apache/beam/pull/37502#discussion_r2784629568
##########
sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilV2.java:
##########
@@ -0,0 +1,330 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.beam.sdk.extensions.gcp.util;
+
+import static org.apache.beam.sdk.io.FileSystemUtils.wildcardToRegexp;
+import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import com.google.api.gax.paging.Page;
+import com.google.auto.value.AutoValue;
+import com.google.cloud.storage.Blob;
+import com.google.cloud.storage.Bucket;
+import com.google.cloud.storage.BucketInfo;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.Storage.BlobField;
+import com.google.cloud.storage.Storage.BlobGetOption;
+import com.google.cloud.storage.Storage.BlobListOption;
+import com.google.cloud.storage.Storage.BucketField;
+import com.google.cloud.storage.Storage.BucketGetOption;
+import com.google.cloud.storage.StorageBatch;
+import com.google.cloud.storage.StorageBatchResult;
+import com.google.cloud.storage.StorageException;
+import com.google.cloud.storage.StorageOptions;
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.nio.file.AccessDeniedException;
+import java.nio.file.FileAlreadyExistsException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.regex.Pattern;
+import org.apache.beam.sdk.extensions.gcp.options.GcpOptions;
+import org.apache.beam.sdk.extensions.gcp.util.gcsfs.GcsPath;
+import org.apache.beam.sdk.options.DefaultValueFactory;
+import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
+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;
+
+class GcsUtilV2 {
+ private static final org.slf4j.Logger LOG = org.slf4j.LoggerFactory.getLogger(GcsUtilV2.class);
+
+ public static class GcsUtilFactory implements DefaultValueFactory<GcsUtilV2> {
+ @Override
+ public GcsUtilV2 create(PipelineOptions options) {
+ // GcsOptions gcsOptions = options.as(GcsOptions.class);
+ // Storage.Builder storageBuilder = Transport.newStorageClient(gcsOptions);
+ return new GcsUtilV2(options);
+ }
+ }
+
+ private Storage storage;
+
+ /** Maximum number of items to retrieve per Objects.List request. */
+ private static final long MAX_LIST_BLOBS_PER_CALL = 1024;
+
+ /** Maximum number of requests permitted in a GCS batch request. */
+ private static final int MAX_REQUESTS_PER_BATCH = 100;
+
+ GcsUtilV2(PipelineOptions options) {
+ String projectId = options.as(GcpOptions.class).getProject();
+ storage = StorageOptions.newBuilder().setProjectId(projectId).build().getService();
+ }
+
+ @SuppressWarnings({
+ "nullness" // For Creating AccessDeniedException and FileAlreadyExistsException with null.
+ })
+ private IOException translateStorageException(GcsPath gcsPath, StorageException e) {
+ switch (e.getCode()) {
+ case 403:
+ return new AccessDeniedException(gcsPath.toString(), null, e.getMessage());
+ case 409:
+ return new FileAlreadyExistsException(gcsPath.toString(), null, e.getMessage());
+ default:
+ return new IOException(e);
+ }
+ }
+
+ private IOException translateStorageException(
+ String bucketName, @Nullable String blobName, StorageException e) {
+ return translateStorageException(GcsPath.fromComponents(bucketName, blobName), e);
+ }
+
+ public Blob getBlob(GcsPath gcsPath, BlobGetOption... options) throws IOException {
+ try {
+ Blob blob = storage.get(gcsPath.getBucket(), gcsPath.getObject(), options);
+ if (blob == null) {
+ throw new FileNotFoundException(
+ String.format("The specified file does not exist: %s", gcsPath.toString()));
+ }
+ return blob;
+ } catch (StorageException e) {
+ throw translateStorageException(gcsPath, e);
+ }
+ }
+
+ public long fileSize(GcsPath gcsPath) throws IOException {
+ return getBlob(gcsPath, BlobGetOption.fields(BlobField.SIZE)).getSize();
+ }
+
+ /** A class that holds either a {@link Blob} or an {@link IOException}. */
+ // It is clear from the name that this class holds either Blob or IOException.
+ @SuppressFBWarnings("NM_CLASS_NOT_EXCEPTION")
+ @AutoValue
+ public abstract static class BlobOrIOException {
+
+ /** Returns the {@link Blob}. */
+ public abstract @Nullable Blob blob();
+
+ /** Returns the {@link IOException}. */
+ public abstract @Nullable IOException ioException();
+
+ @VisibleForTesting
+ public static BlobOrIOException create(Blob blob) {
+ return new AutoValue_GcsUtilV2_BlobOrIOException(
+ checkNotNull(blob, "blob"), null /* ioException */);
+ }
+
+ @VisibleForTesting
+ public static BlobOrIOException create(IOException ioException) {
+ return new AutoValue_GcsUtilV2_BlobOrIOException(
+ null /* blob */, checkNotNull(ioException, "ioException"));
+ }
+ }
+
+ public List<BlobOrIOException> getBlobs(List<GcsPath> gcsPaths, BlobGetOption... options)
+ throws IOException {
+
+ List<BlobOrIOException> results = new ArrayList<>();
+
+ for (List<GcsPath> pathPartition :
+ Lists.partition(Lists.newArrayList(gcsPaths), MAX_REQUESTS_PER_BATCH)) {
+
+ // Create a new empty batch every time
+ StorageBatch batch = storage.batch();
+ List<StorageBatchResult<Blob>> batchResultFutures = new ArrayList<>();
+
+ for (GcsPath path : pathPartition) {
+ batchResultFutures.add(batch.get(path.getBucket(), path.getObject(), options));
+ }
+ batch.submit();
+
+ for (int i = 0; i < batchResultFutures.size(); i++) {
+ StorageBatchResult<Blob> future = batchResultFutures.get(i);
+ try {
+ Blob blob = future.get();
+ if (blob != null) {
+ results.add(BlobOrIOException.create(blob));
+ } else {
+ results.add(
+ BlobOrIOException.create(
+ new FileNotFoundException(
+ String.format(
+ "The specified file does not exist: %s", gcsPaths.get(i).toString()))));
+ }
+ } catch (StorageException e) {
+ results.add(BlobOrIOException.create(translateStorageException(gcsPaths.get(i), e)));
+ }
+ }
+ }
+ return results;
+ }
+
+ /** Lists {@link Blob}s given the {@code bucket}, {@code prefix}, {@code pageToken}. */
+ public Page<Blob> listBlobs(
+ String bucket,
+ String prefix,
+ @Nullable String pageToken,
+ @Nullable String delimiter,
+ BlobListOption... options)
+ throws IOException {
+ List<BlobListOption> blobListOptions = new ArrayList<>();
+ blobListOptions.add(BlobListOption.pageSize(MAX_LIST_BLOBS_PER_CALL));
+ if (pageToken != null) {
+ blobListOptions.add(BlobListOption.pageToken(pageToken));
+ }
+ if (prefix != null) {
+ blobListOptions.add(BlobListOption.prefix(prefix));
+ }
+ if (delimiter != null) {
+ blobListOptions.add(BlobListOption.delimiter(delimiter));
+ }
+ if (options != null && options.length > 0) {
+ for (BlobListOption option : options) {
+ blobListOptions.add(option);
+ }
+ }
+
+ try {
+ return storage.list(bucket, blobListOptions.toArray(new BlobListOption[0]));
+ } catch (StorageException e) {
+ throw translateStorageException(bucket, prefix, e);
+ }
+ }
+
+ public Page<Blob> listBlobs(
+ String bucket, String prefix, @Nullable String pageToken, BlobListOption... options)
+ throws IOException {
+ return listBlobs(bucket, prefix, pageToken, null, options);
+ }
+
+ /**
+ * Expands a pattern into matched paths. The pattern path may contain globs, which are expanded in
+ * the result. For patterns that only match a single object, we ensure that the object exists.
+ */
+ public List<GcsPath> expand(GcsPath gcsPattern) throws IOException {
+ // Handle Non-Wildcard Path
+ if (!GcsPath.isWildcard(gcsPattern)) {
+ try {
+ // Use a get request to fetch the metadata of the object, and ignore the return value.
+ // The request has strong global consistency.
+ getBlob(gcsPattern, BlobGetOption.fields(BlobField.NAME));
+ return ImmutableList.of(gcsPattern);
+ } catch (FileNotFoundException e) {
+ // If the path was not found, return an empty list.
+ return ImmutableList.of();
+ }
+ }
+
+ // Handle Wildcard Path
+ // TODO: check out BlobListOption.matchGlob() for a similar function.
+ String prefix = GcsPath.getNonWildcardPrefix(gcsPattern.getObject());
+ Pattern p = Pattern.compile(wildcardToRegexp(gcsPattern.getObject()));
+
+ LOG.debug(
+ "matching files in bucket {}, prefix {} against pattern {}",
+ gcsPattern.getBucket(),
+ prefix,
+ p.toString());
+
+ List<GcsPath> results = new ArrayList<>();
+ Page<Blob> blobs =
+ listBlobs(
+ gcsPattern.getBucket(),
+ prefix,
+ null,
+ BlobListOption.fields(BlobField.NAME, BlobField.BUCKET));
+ // Iterate through all elements page by page (lazily)
+ for (Blob b : blobs.iterateAll()) {
+ String name = b.getName();
+ // Filter objects based on the regex. Skip directories, which end with a slash.
+ if (p.matcher(name).matches() && !name.endsWith("/")) {
+ LOG.debug("Matched object: {}", name);
+ results.add(GcsPath.fromComponents(b.getBucket(), b.getName()));
+ }
+ }
+ return results;
+ }
+
+ /** Get the {@link Bucket} from Cloud Storage path or propagates an exception. */
+ public Bucket getBucket(GcsPath path, BucketGetOption... options) throws IOException {
+ String bucketName = path.getBucket();
+ try {
+ Bucket bucket = storage.get(bucketName, options);
+ if (bucket == null) {
+ throw new FileNotFoundException(
+ String.format("The specified bucket does not exist: gs://%s", bucketName));
+ }
+ return bucket;
+ } catch (StorageException e) {
+ throw translateStorageException(bucketName, null, e);
+ }
+ }
+
+ /** Returns whether the GCS bucket exists and is accessible. */
+ public boolean bucketAccessible(GcsPath path) {
+ try {
+ // Fetch only the name field to minimize data transfer
+ getBucket(path, BucketGetOption.fields(BucketField.NAME));
+ return true;
+ } catch (IOException e) {
+ return false;
+ }
+ }
+
+ /**
+ * Checks whether the GCS bucket exists. Similar to {@link #bucketAccessible(GcsPath)}, but throws
+ * exception if the bucket is inaccessible due to permissions or does not exist.
+ */
+ public void verifyBucketAccessible(GcsPath path) throws IOException {
+ // Fetch only the name field to minimize data transfer
+ getBucket(path, BucketGetOption.fields(BucketField.NAME));
+ }
+
+ /**
+ * Returns the project number of the project which owns this bucket. If the bucket exists, it must
+ * be accessible otherwise the permissions exception will be propagated. If the bucket does not
+ * exist, an exception will be thrown.
+ */
+ public long bucketProject(GcsPath path) throws IOException {
+ Bucket bucket = getBucket(path, BucketGetOption.fields(BucketField.PROJECT));
+ return bucket.getProject().longValue();
+ }
+
+ public void createBucket(BucketInfo bucketInfo) throws IOException {
+ try {
+ storage.create(bucketInfo);
+ } catch (StorageException e) {
+ throw translateStorageException(bucketInfo.getName(), null, e);
+ }
+ }
+
+ public void removeBucket(BucketInfo bucketInfo) throws IOException {
+ Bucket bucket =
+ getBucket(
+ GcsPath.fromComponents(bucketInfo.getName(), null),
+ BucketGetOption.fields(BucketField.NAME));
+
+ try {
+ bucket.delete();
+ } catch (StorageException e) {
+ throw translateStorageException(bucketInfo.getName(), null, e);
+ }
+ }
Review Comment:
Corrected.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GCS client library migration in Java SDK - part 2 [beam]
Posted by "shunping (via GitHub)" <gi...@apache.org>.
shunping commented on code in PR #37502:
URL: https://github.com/apache/beam/pull/37502#discussion_r2784630677
##########
sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtil.java:
##########
@@ -146,36 +188,86 @@ Integer getUploadBufferSizeBytes() {
}
public long fileSize(GcsPath path) throws IOException {
+ if (delegateV2 != null) {
+ return delegateV2.fileSize(path);
+ }
return delegate.fileSize(path);
}
+ /** @deprecated use {@link #getBlob(GcsPath)}. */
+ @Deprecated
public StorageObject getObject(GcsPath gcsPath) throws IOException {
return delegate.getObject(gcsPath);
}
+ /** @deprecated use {@link #getBlob(GcsPath, BlobGetOption...)}. */
+ @Deprecated
@VisibleForTesting
StorageObject getObject(GcsPath gcsPath, BackOff backoff, Sleeper sleeper) throws IOException {
return delegate.getObject(gcsPath, backoff, sleeper);
}
+ public Blob getBlob(GcsPath gcsPath, BlobGetOption... options) throws IOException {
+ if (delegateV2 != null) {
+ return delegateV2.getBlob(gcsPath, options);
+ }
+ throw new IOException("GcsUtil2 not initialized.");
Review Comment:
Fixed this and other places.
##########
sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilV2.java:
##########
@@ -0,0 +1,330 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.beam.sdk.extensions.gcp.util;
+
+import static org.apache.beam.sdk.io.FileSystemUtils.wildcardToRegexp;
+import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import com.google.api.gax.paging.Page;
+import com.google.auto.value.AutoValue;
+import com.google.cloud.storage.Blob;
+import com.google.cloud.storage.Bucket;
+import com.google.cloud.storage.BucketInfo;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.Storage.BlobField;
+import com.google.cloud.storage.Storage.BlobGetOption;
+import com.google.cloud.storage.Storage.BlobListOption;
+import com.google.cloud.storage.Storage.BucketField;
+import com.google.cloud.storage.Storage.BucketGetOption;
+import com.google.cloud.storage.StorageBatch;
+import com.google.cloud.storage.StorageBatchResult;
+import com.google.cloud.storage.StorageException;
+import com.google.cloud.storage.StorageOptions;
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.nio.file.AccessDeniedException;
+import java.nio.file.FileAlreadyExistsException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.regex.Pattern;
+import org.apache.beam.sdk.extensions.gcp.options.GcpOptions;
+import org.apache.beam.sdk.extensions.gcp.util.gcsfs.GcsPath;
+import org.apache.beam.sdk.options.DefaultValueFactory;
+import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
+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;
+
+class GcsUtilV2 {
+ private static final org.slf4j.Logger LOG = org.slf4j.LoggerFactory.getLogger(GcsUtilV2.class);
+
+ public static class GcsUtilFactory implements DefaultValueFactory<GcsUtilV2> {
+ @Override
+ public GcsUtilV2 create(PipelineOptions options) {
+ // GcsOptions gcsOptions = options.as(GcsOptions.class);
+ // Storage.Builder storageBuilder = Transport.newStorageClient(gcsOptions);
+ return new GcsUtilV2(options);
+ }
+ }
+
+ private Storage storage;
+
+ /** Maximum number of items to retrieve per Objects.List request. */
+ private static final long MAX_LIST_BLOBS_PER_CALL = 1024;
+
+ /** Maximum number of requests permitted in a GCS batch request. */
+ private static final int MAX_REQUESTS_PER_BATCH = 100;
+
+ GcsUtilV2(PipelineOptions options) {
+ String projectId = options.as(GcpOptions.class).getProject();
+ storage = StorageOptions.newBuilder().setProjectId(projectId).build().getService();
+ }
+
+ @SuppressWarnings({
+ "nullness" // For Creating AccessDeniedException and FileAlreadyExistsException with null.
+ })
+ private IOException translateStorageException(GcsPath gcsPath, StorageException e) {
+ switch (e.getCode()) {
+ case 403:
+ return new AccessDeniedException(gcsPath.toString(), null, e.getMessage());
+ case 409:
+ return new FileAlreadyExistsException(gcsPath.toString(), null, e.getMessage());
+ default:
+ return new IOException(e);
+ }
+ }
+
+ private IOException translateStorageException(
+ String bucketName, @Nullable String blobName, StorageException e) {
+ return translateStorageException(GcsPath.fromComponents(bucketName, blobName), e);
+ }
+
+ public Blob getBlob(GcsPath gcsPath, BlobGetOption... options) throws IOException {
+ try {
+ Blob blob = storage.get(gcsPath.getBucket(), gcsPath.getObject(), options);
+ if (blob == null) {
+ throw new FileNotFoundException(
+ String.format("The specified file does not exist: %s", gcsPath.toString()));
+ }
+ return blob;
+ } catch (StorageException e) {
+ throw translateStorageException(gcsPath, e);
+ }
+ }
+
+ public long fileSize(GcsPath gcsPath) throws IOException {
+ return getBlob(gcsPath, BlobGetOption.fields(BlobField.SIZE)).getSize();
+ }
+
+ /** A class that holds either a {@link Blob} or an {@link IOException}. */
+ // It is clear from the name that this class holds either Blob or IOException.
+ @SuppressFBWarnings("NM_CLASS_NOT_EXCEPTION")
+ @AutoValue
+ public abstract static class BlobOrIOException {
+
+ /** Returns the {@link Blob}. */
+ public abstract @Nullable Blob blob();
+
+ /** Returns the {@link IOException}. */
+ public abstract @Nullable IOException ioException();
+
+ @VisibleForTesting
+ public static BlobOrIOException create(Blob blob) {
+ return new AutoValue_GcsUtilV2_BlobOrIOException(
+ checkNotNull(blob, "blob"), null /* ioException */);
+ }
+
+ @VisibleForTesting
+ public static BlobOrIOException create(IOException ioException) {
+ return new AutoValue_GcsUtilV2_BlobOrIOException(
+ null /* blob */, checkNotNull(ioException, "ioException"));
+ }
+ }
+
+ public List<BlobOrIOException> getBlobs(List<GcsPath> gcsPaths, BlobGetOption... options)
+ throws IOException {
+
+ List<BlobOrIOException> results = new ArrayList<>();
+
+ for (List<GcsPath> pathPartition :
+ Lists.partition(Lists.newArrayList(gcsPaths), MAX_REQUESTS_PER_BATCH)) {
+
+ // Create a new empty batch every time
+ StorageBatch batch = storage.batch();
+ List<StorageBatchResult<Blob>> batchResultFutures = new ArrayList<>();
+
+ for (GcsPath path : pathPartition) {
+ batchResultFutures.add(batch.get(path.getBucket(), path.getObject(), options));
+ }
+ batch.submit();
+
+ for (int i = 0; i < batchResultFutures.size(); i++) {
+ StorageBatchResult<Blob> future = batchResultFutures.get(i);
+ try {
+ Blob blob = future.get();
+ if (blob != null) {
+ results.add(BlobOrIOException.create(blob));
+ } else {
+ results.add(
+ BlobOrIOException.create(
+ new FileNotFoundException(
+ String.format(
+ "The specified file does not exist: %s", gcsPaths.get(i).toString()))));
+ }
+ } catch (StorageException e) {
+ results.add(BlobOrIOException.create(translateStorageException(gcsPaths.get(i), e)));
+ }
+ }
Review Comment:
Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GCS client library migration in Java SDK - part 2 [beam]
Posted by "shunping (via GitHub)" <gi...@apache.org>.
shunping commented on code in PR #37502:
URL: https://github.com/apache/beam/pull/37502#discussion_r2795112055
##########
sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilV2.java:
##########
@@ -0,0 +1,329 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.beam.sdk.extensions.gcp.util;
+
+import static org.apache.beam.sdk.io.FileSystemUtils.wildcardToRegexp;
+import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import com.google.api.gax.paging.Page;
+import com.google.auto.value.AutoValue;
+import com.google.cloud.storage.Blob;
+import com.google.cloud.storage.Bucket;
+import com.google.cloud.storage.BucketInfo;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.Storage.BlobField;
+import com.google.cloud.storage.Storage.BlobGetOption;
+import com.google.cloud.storage.Storage.BlobListOption;
+import com.google.cloud.storage.Storage.BucketField;
+import com.google.cloud.storage.Storage.BucketGetOption;
+import com.google.cloud.storage.StorageBatch;
+import com.google.cloud.storage.StorageBatchResult;
+import com.google.cloud.storage.StorageException;
+import com.google.cloud.storage.StorageOptions;
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.nio.file.AccessDeniedException;
+import java.nio.file.FileAlreadyExistsException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.regex.Pattern;
+import org.apache.beam.sdk.extensions.gcp.options.GcpOptions;
+import org.apache.beam.sdk.extensions.gcp.util.gcsfs.GcsPath;
+import org.apache.beam.sdk.options.DefaultValueFactory;
+import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
+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;
+
+class GcsUtilV2 {
+ private static final org.slf4j.Logger LOG = org.slf4j.LoggerFactory.getLogger(GcsUtilV2.class);
+
+ public static class GcsUtilFactory implements DefaultValueFactory<GcsUtilV2> {
+ @Override
+ public GcsUtilV2 create(PipelineOptions options) {
+ // GcsOptions gcsOptions = options.as(GcsOptions.class);
+ // Storage.Builder storageBuilder = Transport.newStorageClient(gcsOptions);
+ return new GcsUtilV2(options);
+ }
+ }
+
+ private Storage storage;
+
+ /** Maximum number of items to retrieve per Objects.List request. */
+ private static final long MAX_LIST_BLOBS_PER_CALL = 1024;
+
+ /** Maximum number of requests permitted in a GCS batch request. */
+ private static final int MAX_REQUESTS_PER_BATCH = 100;
+
+ GcsUtilV2(PipelineOptions options) {
+ String projectId = options.as(GcpOptions.class).getProject();
+ storage = StorageOptions.newBuilder().setProjectId(projectId).build().getService();
+ }
+
+ @SuppressWarnings({
+ "nullness" // For Creating AccessDeniedException and FileAlreadyExistsException with null.
+ })
+ private IOException translateStorageException(GcsPath gcsPath, StorageException e) {
+ switch (e.getCode()) {
+ case 403:
+ return new AccessDeniedException(gcsPath.toString(), null, e.getMessage());
+ case 409:
+ return new FileAlreadyExistsException(gcsPath.toString(), null, e.getMessage());
+ default:
+ return new IOException(e);
+ }
+ }
+
+ private IOException translateStorageException(
+ String bucketName, @Nullable String blobName, StorageException e) {
+ return translateStorageException(GcsPath.fromComponents(bucketName, blobName), e);
+ }
+
+ public Blob getBlob(GcsPath gcsPath, BlobGetOption... options) throws IOException {
+ try {
+ Blob blob = storage.get(gcsPath.getBucket(), gcsPath.getObject(), options);
+ if (blob == null) {
+ throw new FileNotFoundException(
+ String.format("The specified file does not exist: %s", gcsPath.toString()));
+ }
+ return blob;
+ } catch (StorageException e) {
+ throw translateStorageException(gcsPath, e);
+ }
+ }
+
+ public long fileSize(GcsPath gcsPath) throws IOException {
+ return getBlob(gcsPath, BlobGetOption.fields(BlobField.SIZE)).getSize();
+ }
+
+ /** A class that holds either a {@link Blob} or an {@link IOException}. */
+ // It is clear from the name that this class holds either Blob or IOException.
+ @SuppressFBWarnings("NM_CLASS_NOT_EXCEPTION")
Review Comment:
I followed a previous class in GcsUtilV1.java, i.e. `StorageObjectOrIOException`.
https://github.com/apache/beam/blob/aaecfa8f54583ccd227189bd2e8580dcee78f502/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtil.java#L366
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GCS client library migration in Java SDK - part 2 [beam]
Posted by "shunping (via GitHub)" <gi...@apache.org>.
shunping commented on code in PR #37502:
URL: https://github.com/apache/beam/pull/37502#discussion_r2795135522
##########
sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/gcsfs/GcsPathTest.java:
##########
@@ -348,4 +348,24 @@ public void testSubPathError() {
a.subpath(1, 1); // throws IllegalArgumentException
Assert.fail();
}
+
+ @Test
+ public void testIsWildcard() {
+ assertTrue(GcsPath.isWildcard(GcsPath.fromUri("gs://bucket/foo*")));
+ assertTrue(GcsPath.isWildcard(GcsPath.fromUri("gs://bucket/foo?")));
+ assertTrue(GcsPath.isWildcard(GcsPath.fromUri("gs://bucket/foo[a-z]")));
Review Comment:
That regex wildcard is supported by V1 already. I am just adding more tests here and there is no behavior change.
`isWildcard()` is only moved from `GcsUtil` into `GcsPath`, so it can be shared by GcsUtil V1 and V2.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GCS client library migration in Java SDK - part 2 [beam]
Posted by "shunping (via GitHub)" <gi...@apache.org>.
shunping merged PR #37502:
URL: https://github.com/apache/beam/pull/37502
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GCS client library migration in Java SDK - part 2 [beam]
Posted by "shunping (via GitHub)" <gi...@apache.org>.
shunping commented on code in PR #37502:
URL: https://github.com/apache/beam/pull/37502#discussion_r2795135522
##########
sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/gcsfs/GcsPathTest.java:
##########
@@ -348,4 +348,24 @@ public void testSubPathError() {
a.subpath(1, 1); // throws IllegalArgumentException
Assert.fail();
}
+
+ @Test
+ public void testIsWildcard() {
+ assertTrue(GcsPath.isWildcard(GcsPath.fromUri("gs://bucket/foo*")));
+ assertTrue(GcsPath.isWildcard(GcsPath.fromUri("gs://bucket/foo?")));
+ assertTrue(GcsPath.isWildcard(GcsPath.fromUri("gs://bucket/foo[a-z]")));
Review Comment:
That regex wildcard is supported by V1 already. I am just adding more test coverage here and there is no behavior change.
`isWildcard()` is only moved from `GcsUtil` into `GcsPath`, so it can be shared by GcsUtil V1 and V2.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GCS client library migration in Java SDK - part 2 [beam]
Posted by "shunping (via GitHub)" <gi...@apache.org>.
shunping commented on code in PR #37502:
URL: https://github.com/apache/beam/pull/37502#discussion_r2795135522
##########
sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/gcsfs/GcsPathTest.java:
##########
@@ -348,4 +348,24 @@ public void testSubPathError() {
a.subpath(1, 1); // throws IllegalArgumentException
Assert.fail();
}
+
+ @Test
+ public void testIsWildcard() {
+ assertTrue(GcsPath.isWildcard(GcsPath.fromUri("gs://bucket/foo*")));
+ assertTrue(GcsPath.isWildcard(GcsPath.fromUri("gs://bucket/foo?")));
+ assertTrue(GcsPath.isWildcard(GcsPath.fromUri("gs://bucket/foo[a-z]")));
Review Comment:
No behavior change here. I am just adding more tests here.
`isWildcard()` is only moved from `GcsUtil` into `GcsPath`, so it can be shared by GcsUtil V1 and V2.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GCS client library migration in Java SDK - part 2 [beam]
Posted by "shunping (via GitHub)" <gi...@apache.org>.
shunping commented on PR #37502:
URL: https://github.com/apache/beam/pull/37502#issuecomment-3873389639
R: @Abacn
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org