[Java] Use GcsUtil V2 by default

.github/trigger_files/beam_PostCommit_Java.json

@@ -1,5 +1,5 @@
 {
   "https://github.com/apache/beam/pull/39893": "Fix nullness in BigQueryIO",
   "comment": "Modify this file in a trivial way to cause this test suite to run",
-  "modification": 6
+  "modification": 7
 } 

.github/trigger_files/beam_PostCommit_Java_DataflowV1.json

@@ -4,6 +4,6 @@
   "https://github.com/apache/beam/pull/34902": "Introducing OutputBuilder",
   "https://github.com/apache/beam/pull/35177": "Introducing WindowedValueReceiver to runners",
     "comment": "Modify this file in a trivial way to cause this test suite to run",
-    "modification": 9,
+    "modification": 10,
   "https://github.com/apache/beam/pull/35159": "moving WindowedValue and making an interface"
 }

.github/trigger_files/beam_PostCommit_Java_DataflowV2.json

@@ -1,6 +1,6 @@
 {
   "https://github.com/apache/beam/pull/39893": "Fix nullness in BigQueryIO",
-  "modification": 9,
+  "modification": 10,
   "https://github.com/apache/beam/pull/35159": "moving WindowedValue and making an interface",
   "https://github.com/apache/beam/pull/39990": "removing dead code from FnApiDoFnRunner"
 }

.github/trigger_files/beam_PostCommit_Java_Examples_Dataflow.json

@@ -0,0 +1,4 @@
+{
+  "comment": "Modify this file in a trivial way to cause this test suite to run",
+  "modification": 1
+}

.github/trigger_files/beam_PostCommit_Java_Examples_Dataflow_V2.json

@@ -2,6 +2,6 @@
   "https://github.com/apache/beam/pull/36138": "Cleanly separating v1 worker and v2 sdk harness container image handling",
   "https://github.com/apache/beam/pull/34902": "Introducing OutputBuilder",
   "comment": "Modify this file in a trivial way to cause this test suite to run",
-  "modification": 2,
+  "modification": 3,
   "https://github.com/apache/beam/pull/39990": "removing dead code from FnApiDoFnRunner"
 }

.github/trigger_files/beam_PostCommit_Java_Examples_Direct.json

@@ -0,0 +1,4 @@
+{
+  "comment": "Modify this file in a trivial way to cause this test suite to run",
+  "modification": 1
+}

.github/trigger_files/beam_PostCommit_Java_Hadoop_Versions.json

@@ -1,4 +1,4 @@
 {
   "comment": "Modify this file in a trivial way to cause this test suite to run",
-  "modification": 5
+  "modification": 6
 }

.github/trigger_files/beam_PostCommit_Java_IO_Performance_Tests.json

@@ -1,4 +1,4 @@
 {
   "comment": "Modify this file in a trivial way to cause this test suite to run",
-  "modification": 4,
+  "modification": 5,
 }

CHANGES.md

@@ -84,6 +84,7 @@
 * (Go) Added `schema.RegisterLogicalTypeConversion` and `schema.RegisterLogicalTypeCoder` to register portable logical types in one call, and Go types for the standard `beam:logical_type:date:v1` and `beam:logical_type:micros_instant:v1` logical types, so schemas from other SDKs with these types can be used in cross-language pipelines ([#39684](https://github.com/apache/beam/issues/39684)).
 * (Python, Java, Go) Added `--dataflow_container_base_repository` (Python, Go) and `--dataflowContainerBaseRepository` (Java) to set the repository (e.g. `docker.example.com/my-prefix/cloud-dataflow/v1beta3`) from which the Dataflow runner pulls its default SDK container images, including the default images of cross-language Apache Beam environments. Defaults to `gcr.io/cloud-dataflow/v1beta3` ([#40569](https://github.com/apache/beam/issues/40569)).
 * (Go) GCS and Dataflow clients now support non-default universes such as Trusted Partner Cloud (TPC), using the universe domain from `GOOGLE_CLOUD_UNIVERSE_DOMAIN` or the default credentials. Added a `--gcs_endpoint` pipeline option to override the GCS endpoint ([#40569](https://github.com/apache/beam/issues/40569)).
+* (Java) `GcsUtil` now uses the java-storage client (GcsUtil V2) for GCS operations by default instead of gcsio (GcsUtil V1). To go back to V1, pass `--experiments=disable_gcsutil_v2`. Pipelines with the `use_grpc_for_gcs` experiment keep using V1. The `use_gcsutil_v2` experiment is no longer needed and is now ignored ([#40712](https://github.com/apache/beam/issues/40712)).
 
 ## Breaking Changes
 

it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/storage/ParquetIOLT.java

@@ -132,15 +132,17 @@
  * <h3>GcsUtil v1 or v2</h3>
  *
  * <p>{@code useGcsUtilV2} routes GCS access through the java-storage client and defaults to off,
- * i.e. to the gcsio based GcsUtilV1 that every pipeline uses today:
+ * i.e. to the gcsio based GcsUtilV1:
  *
  * <pre>
  * -Dconfiguration='{"preset":"f100_s16","useGcsUtilV2":true}'
  * </pre>
  *
- * <p>On Dataflow this is sent as the {@code use_gcsutil_v2} experiment, for the other runners it is
- * set on the pipeline options directly. Either way {@code GcsUtil} logs the version it selected at
- * INFO, which is what to grep for to confirm a run really used the intended client.
+ * <p>The version is always selected explicitly, with the {@code use_gcsutil_v2} or the {@code
+ * disable_gcsutil_v2} experiment, so a run does not depend on which one {@code GcsUtil} defaults
+ * to. On Dataflow the experiment is sent as a job parameter, for the other runners it is set on the
+ * pipeline options directly. {@code GcsUtil} logs the version it selected at INFO, which is what to
+ * grep for to confirm a run really used the intended client.
  *
  * <p>The number of requests a write costs is decided by the upload chunk size, which {@code
  * gcsUploadBufferSizeBytes} pins when a run needs to sweep it or to rule it out as a variable:
@@ -207,6 +209,9 @@ public final class ParquetIOLT extends GcsIOLoadTestBase {
   /** Experiment that routes GCS access through GcsUtilV2, i.e. the java-storage client. */
   private static final String GCS_UTIL_V2_EXPERIMENT = "use_gcsutil_v2";
 
+  /** Experiment that falls back to GcsUtilV1, i.e. the gcsio client. */
+  private static final String GCS_UTIL_DISABLE_V2_EXPERIMENT = "disable_gcsutil_v2";
+
   /** Pipeline option that pins the upload chunk size, see {@code GcsOptions}. */
   private static final String GCS_UPLOAD_BUFFER_SIZE_OPTION = "gcsUploadBufferSizeBytes";
 
@@ -467,10 +472,8 @@ private PipelineLauncher.LaunchConfig launchConfig(String jobName, TestPipeline
     pipeline.getOptions().as(GcsOptions.class).setGcsPerformanceMetrics(true);
     // Same reason: for a non-Dataflow runner the experiments parameter below never reaches the
     // pipeline, so the experiment has to be added to the options directly.
-    if (configuration.useGcsUtilV2) {
-      ExperimentalOptions.addExperiment(
-          pipeline.getOptions().as(ExperimentalOptions.class), GCS_UTIL_V2_EXPERIMENT);
-    }
+    ExperimentalOptions.addExperiment(
+        pipeline.getOptions().as(ExperimentalOptions.class), gcsUtilExperiment());
     // Left unset the two clients pick different chunk sizes, which shows up as a difference in the
     // number of write requests.
     if (configuration.gcsUploadBufferSizeBytes > 0) {
@@ -539,12 +542,18 @@ private PipelineLauncher.LaunchConfig launchConfig(String jobName, TestPipeline
   private static String dataflowExperiments() {
     List<String> experiments = new ArrayList<>();
     experiments.add(configuration.useRunnerV2 ? RUNNER_V2_EXPERIMENT : LEGACY_WORKER_EXPERIMENT);
-    if (configuration.useGcsUtilV2) {
-      experiments.add(GCS_UTIL_V2_EXPERIMENT);
-    }
+    experiments.add(gcsUtilExperiment());
     return String.join(",", experiments);
   }
 
+  /**
+   * The experiment that selects the GcsUtil version under test. Both versions are named explicitly,
+   * so that a run is not affected by which one {@code GcsUtil} picks by default.
+   */
+  private static String gcsUtilExperiment() {
+    return configuration.useGcsUtilV2 ? GCS_UTIL_V2_EXPERIMENT : GCS_UTIL_DISABLE_V2_EXPERIMENT;
+  }
+
   /**
    * Resolves {@code -Dconfiguration} into a {@link Configuration}.
    *
@@ -892,11 +901,12 @@ static class Configuration extends SyntheticSourceOptions {
     @JsonProperty public boolean useRunnerV2 = true;
 
     /**
-     * {@code true} routes GCS access through GcsUtilV2, the java-storage client, instead of the
-     * default GcsUtilV1. Applies to every runner: on Dataflow it is sent as the {@code
-     * use_gcsutil_v2} experiment, elsewhere it is set on the pipeline options directly.
+     * {@code true} routes GCS access through GcsUtilV2, the java-storage client, {@code false}
+     * through GcsUtilV1, the gcsio client. Applies to every runner: on Dataflow it is sent as the
+     * {@code use_gcsutil_v2} or {@code disable_gcsutil_v2} experiment, elsewhere it is set on the
+     * pipeline options directly.
      *
-     * <p>Which one a run actually used is logged by {@code GcsUtil} at INFO.
+     * <p>{@code GcsUtil} logs the version a run actually used at INFO.
      */
     @JsonProperty public boolean useGcsUtilV2 = false;
 

sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/options/GcsOptions.java

@@ -174,15 +174,15 @@ public GoogleCloudStorageReadOptions create(PipelineOptions options) {
   /**
    * Batching limit for rewrite ops which will copy data.
    *
-   * @deprecated Only applicable to GcsUtil V1. Not applicable to GcsUtil V2 (enabled with the
-   *     {@code use_gcsutil_v2} experiment), which does not batch rewrites; use {@link
-   *     #getGcsMaxConcurrentRewrites} to tune copy/move concurrency there instead.
+   * @deprecated Only applicable to GcsUtil V1 (selected with the {@code disable_gcsutil_v2}
+   *     experiment). Not applicable to GcsUtil V2 (the default), which does not batch rewrites; use
+   *     {@link #getGcsMaxConcurrentRewrites} to tune copy/move concurrency there instead.
    */
   @Deprecated
   @Description(
-      "Batching limit for rewrite ops which will copy data. Deprecated: NOT applicable to GcsUtil"
-          + " V2 (enabled with the use_gcsutil_v2 experiment); use gcsMaxConcurrentRewrites"
-          + " instead.")
+      "Batching limit for rewrite ops which will copy data. Deprecated: only applicable to"
+          + " GcsUtil V1 (selected with the disable_gcsutil_v2 experiment), NOT to GcsUtil V2 (the"
+          + " default); use gcsMaxConcurrentRewrites instead.")
   @Nullable Integer getGcsRewriteDataOpBatchLimit();
 
   /**
@@ -195,12 +195,13 @@ public GoogleCloudStorageReadOptions create(PipelineOptions options) {
    * Maximum number of files rewritten concurrently when copying or moving files. If unset, a
    * default of 128 is used.
    *
-   * <p>Only applicable to GcsUtil V2, which is enabled with the {@code use_gcsutil_v2} experiment.
+   * <p>Only applicable to GcsUtil V2, which is the default. Ignored when falling back to GcsUtil V1
+   * with the {@code disable_gcsutil_v2} experiment.
    */
   @Description(
       "Maximum number of files rewritten concurrently when copying or moving files in GCS."
-          + " If unset, defaults to 128. Only applicable to GcsUtil V2 (enabled with the"
-          + " use_gcsutil_v2 experiment).")
+          + " If unset, defaults to 128. Only applicable to GcsUtil V2 (the default); ignored"
+          + " with the disable_gcsutil_v2 experiment.")
   @Nullable Integer getGcsMaxConcurrentRewrites();
 
   void setGcsMaxConcurrentRewrites(@Nullable Integer maxConcurrentRewrites);

sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtil.java

@@ -127,16 +127,41 @@ public static boolean isWildcard(GcsPath spec) {
     return GcsPath.isWildcard(spec);
   }
 
+  /**
+   * Experiment that falls back to GcsUtil V1 (gcsio). V2 (java-storage) is the default, and this
+   * experiment is the escape hatch should V2 misbehave for a pipeline.
+   */
+  public static final String DISABLE_GCSUTIL_V2_EXPERIMENT = "disable_gcsutil_v2";
+
+  /**
+   * Experiment that used to opt into GcsUtil V2. V2 is now the default, so this is a no-op kept so
+   * that existing pipelines that still pass it keep working. {@link #DISABLE_GCSUTIL_V2_EXPERIMENT}
+   * takes precedence if both are set.
+   */
+  public static final String USE_GCSUTIL_V2_EXPERIMENT = "use_gcsutil_v2";
+
+  /**
+   * Experiment that makes GcsUtil V1 talk to GCS over gRPC. Only V1 (gcsio) supports it, so it also
+   * falls back to V1, the same as {@link #DISABLE_GCSUTIL_V2_EXPERIMENT}.
+   */
+  public static final String USE_GRPC_FOR_GCS_EXPERIMENT = "use_grpc_for_gcs";
+
   GcsUtil(PipelineOptions options) {
     this.delegate = new GcsUtilV1.GcsUtilFactory().create(options);
-    if (ExperimentalOptions.hasExperiment(options, "use_gcsutil_v2")) {
+    @Nullable String v1Experiment = null;
+    if (ExperimentalOptions.hasExperiment(options, DISABLE_GCSUTIL_V2_EXPERIMENT)) {
+      v1Experiment = DISABLE_GCSUTIL_V2_EXPERIMENT;
+    } else if (ExperimentalOptions.hasExperiment(options, USE_GRPC_FOR_GCS_EXPERIMENT)) {
+      v1Experiment = USE_GRPC_FOR_GCS_EXPERIMENT;
+    }
+    if (v1Experiment != null) {
+      this.delegateV2 = null;
+      LOG.info(
+          "Using GcsUtilV1 (gcsio) for GCS operations because the {} experiment is set.",
+          v1Experiment);
+    } else {
       this.delegateV2 = new GcsUtilV2.GcsUtilFactory().create(options);
-      // INFO only for V2, which is opt-in. V1 is still the default for every pipeline,
-      // so logging it at INFO would be noise.
       LOG.info("Using GcsUtilV2 (java-storage) for GCS operations.");
-    } else {
-      this.delegateV2 = null;
-      LOG.debug("Using GcsUtilV1 (gcsio) for GCS operations.");
     }
   }
 
@@ -633,7 +658,8 @@ private static StorageObject toStorageObject(Blob blob) {
             .setContentEncoding(blob.getContentEncoding())
             .setMd5Hash(blob.getMd5())
             .setCrc32c(blob.getCrc32c())
-            .setEtag(blob.getEtag());
+            .setEtag(blob.getEtag())
+            .setKmsKeyName(blob.getKmsKeyName());
     Long size = blob.getSize();
     if (size != null) {
       storageObject.setSize(BigInteger.valueOf(size));

sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilV1.java

@@ -171,7 +171,7 @@ public GcsUtilV1 create(PipelineOptions options) {
           storageBuilder.build(),
           storageBuilder.getHttpRequestInitializer(),
           gcsOptions.getExecutorService(),
-          hasExperiment(options, "use_grpc_for_gcs"),
+          hasExperiment(options, GcsUtil.USE_GRPC_FOR_GCS_EXPERIMENT),
           gcsOptions.getGcpCredential(),
           gcsOptions.getGcsUploadBufferSizeBytes(),
           gcsOptions.getGcsRewriteDataOpBatchLimit(),

sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilIT.java

@@ -109,7 +109,7 @@ public class GcsUtilIT {
 
   @Parameters(name = "{0}")
   public static Iterable<String> data() {
-    return Arrays.asList("use_gcsutil_v1", "use_gcsutil_v2");
+    return Arrays.asList("disable_gcsutil_v2", "use_gcsutil_v2");
   }
 
   @Parameter public String experiment;
@@ -1342,6 +1342,9 @@ public void testRewriteMultiPart() throws IOException {
 
     TestPipelineOptions options =
         TestPipeline.testingPipelineOptions().as(TestPipelineOptions.class);
+    // These options are built from scratch, so the experiment set in setUp() has to be carried
+    // over, otherwise GcsUtil picks its default (V2) rather than the V1 this test exercises.
+    options.as(ExperimentalOptions.class).setExperiments(Collections.singletonList(experiment));
     // Using a KMS key is necessary to trigger multi-part rewrites (bucket is created
     // with a bucket default key).
     assertNotNull(options.getTempRoot());

sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilTest.java

@@ -43,6 +43,7 @@
 import java.time.Duration;
 import java.time.Instant;
 import java.time.ZoneOffset;
+import java.util.Arrays;
 import java.util.List;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
@@ -52,6 +53,7 @@
 import org.apache.beam.sdk.extensions.gcp.util.GcsUtil.StorageObjectOrIOException;
 import org.apache.beam.sdk.extensions.gcp.util.gcsfs.GcsPath;
 import org.apache.beam.sdk.io.fs.MoveOptions.StandardMoveOptions;
+import org.apache.beam.sdk.options.ExperimentalOptions;
 import org.apache.beam.sdk.options.PipelineOptionsFactory;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList;
 import org.junit.Test;
@@ -117,7 +119,7 @@ public void testMultipleThreadsCanCompleteOutOfOrderWithDefaultThreadPool() thro
   }
 
   // The tests below cover the routing that the GcsUtil facade performs for the legacy typed
-  // methods once the use_gcsutil_v2 experiment installs a GcsUtilV2 delegate. Both delegates are
+  // methods when a GcsUtilV2 delegate is installed (the default). Both delegates are
   // mocked, so they assert both that V2 is used and that V1 is left alone.
 
   private GcsUtilV1 mockDelegate;
@@ -333,6 +335,7 @@ public void testGetObjectIsRoutedToV2AndKeepsAllFields() throws IOException {
     when(blob.getMd5()).thenReturn("md5==");
     when(blob.getCrc32c()).thenReturn("crc==");
     when(blob.getEtag()).thenReturn("etag");
+    when(blob.getKmsKeyName()).thenReturn("projects/p/locations/l/keyRings/r/cryptoKeys/k");
     when(blob.getUpdateTimeOffsetDateTime())
         .thenReturn(Instant.ofEpochMilli(1234L).atOffset(ZoneOffset.UTC));
     when(blob.getCreateTimeOffsetDateTime())
@@ -351,6 +354,7 @@ public void testGetObjectIsRoutedToV2AndKeepsAllFields() throws IOException {
     assertEquals("md5==", object.getMd5Hash());
     assertEquals("crc==", object.getCrc32c());
     assertEquals("etag", object.getEtag());
+    assertEquals("projects/p/locations/l/keyRings/r/cryptoKeys/k", object.getKmsKeyName());
     assertEquals(1234L, object.getUpdated().getValue());
     assertEquals(1000L, object.getTimeCreated().getValue());
     Mockito.verifyNoMoreInteractions(mockDelegate);
@@ -439,10 +443,54 @@ public void testGetBucketWithoutSoftDeletePolicyForV2() throws IOException {
     assertNull(converted.getStorageClass());
   }
 
-  /** Without the use_gcsutil_v2 experiment, the V2-only methods fail rather than fall back. */
+  private static GcsOptions gcsOptionsWithExperiments(String... experiments) {
+    GcsOptions options = gcsOptionsWithTestCredential();
+    options.as(ExperimentalOptions.class).setExperiments(Arrays.asList(experiments));
+    return options;
+  }
+
+  /** GcsUtil V2 is used by default. */
   @Test
-  public void testV2OnlyMethodsFailWithoutV2() {
+  public void testUsesV2ByDefault() {
     GcsUtil gcsUtil = gcsOptionsWithTestCredential().getGcsUtil();
+    assertNotNull(gcsUtil.delegateV2);
+  }
+
+  /** The legacy opt-in experiment is still accepted and keeps V2. */
+  @Test
+  public void testUsesV2WithLegacyV2Experiment() {
+    GcsUtil gcsUtil = gcsOptionsWithExperiments(GcsUtil.USE_GCSUTIL_V2_EXPERIMENT).getGcsUtil();
+    assertNotNull(gcsUtil.delegateV2);
+  }
+
+  /** The disable_gcsutil_v2 experiment falls back to V1, even when use_gcsutil_v2 is also set. */
+  @Test
+  public void testFallsBackToV1WithDisableV2Experiment() {
+    assertNull(
+        gcsOptionsWithExperiments(GcsUtil.DISABLE_GCSUTIL_V2_EXPERIMENT).getGcsUtil().delegateV2);
+    assertNull(
+        gcsOptionsWithExperiments(
+                GcsUtil.DISABLE_GCSUTIL_V2_EXPERIMENT, GcsUtil.USE_GCSUTIL_V2_EXPERIMENT)
+            .getGcsUtil()
+            .delegateV2);
+  }
+
+  /** gRPC is only supported by V1, so the use_grpc_for_gcs experiment falls back to V1. */
+  @Test
+  public void testFallsBackToV1WithGrpcExperiment() {
+    assertNull(
+        gcsOptionsWithExperiments(GcsUtil.USE_GRPC_FOR_GCS_EXPERIMENT).getGcsUtil().delegateV2);
+    assertNull(
+        gcsOptionsWithExperiments(
+                GcsUtil.USE_GRPC_FOR_GCS_EXPERIMENT, GcsUtil.USE_GCSUTIL_V2_EXPERIMENT)
+            .getGcsUtil()
+            .delegateV2);
+  }
+
+  /** With the disable_gcsutil_v2 experiment, the V2-only methods fail rather than fall back. */
+  @Test
+  public void testV2OnlyMethodsFailWithoutV2() {
+    GcsUtil gcsUtil = gcsOptionsWithExperiments(GcsUtil.DISABLE_GCSUTIL_V2_EXPERIMENT).getGcsUtil();
     assertNull(gcsUtil.delegateV2);
     GcsUtilV1 v1 = Mockito.mock(GcsUtilV1.class);
     gcsUtil.delegate = v1;