From 05d320ddc7013582f81b3c6219bd417e423012a6 Mon Sep 17 00:00:00 2001 From: Shunping Huang Date: Wed, 21 Oct 2026 14:35:19 -0400 Subject: [PATCH 1/7] Use GcsUtil V2 by default, add disable_gcsutil_v2 fallback GcsUtil now uses GcsUtilV2 (java-storage) by default. The disable_gcsutil_v2 experiment falls back to GcsUtilV1 (gcsio). use_gcsutil_v2 is still accepted but does nothing. --- .../extensions/gcp/options/GcsOptions.java | 19 ++++----- .../beam/sdk/extensions/gcp/util/GcsUtil.java | 25 +++++++++--- .../sdk/extensions/gcp/util/GcsUtilIT.java | 2 +- .../sdk/extensions/gcp/util/GcsUtilTest.java | 40 +++++++++++++++++-- .../extensions/gcp/util/GcsUtilV1Test.java | 23 ++++++++--- 5 files changed, 85 insertions(+), 24 deletions(-) diff --git a/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/options/GcsOptions.java b/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/options/GcsOptions.java index b5246a87d2ba..88ebf73dfab7 100644 --- a/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/options/GcsOptions.java +++ b/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. * - *

Only applicable to GcsUtil V2, which is enabled with the {@code use_gcsutil_v2} experiment. + *

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); diff --git a/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtil.java b/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtil.java index 382ac6856bf4..e9e34e3cc2b7 100644 --- a/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtil.java +++ b/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtil.java @@ -127,16 +127,29 @@ 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"; + GcsUtil(PipelineOptions options) { this.delegate = new GcsUtilV1.GcsUtilFactory().create(options); - if (ExperimentalOptions.hasExperiment(options, "use_gcsutil_v2")) { + if (ExperimentalOptions.hasExperiment(options, DISABLE_GCSUTIL_V2_EXPERIMENT)) { + this.delegateV2 = null; + LOG.info( + "Using GcsUtilV1 (gcsio) for GCS operations because the {} experiment is set.", + DISABLE_GCSUTIL_V2_EXPERIMENT); + } 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."); } } diff --git a/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilIT.java b/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilIT.java index a6ae12024a10..423c40fe8b4b 100644 --- a/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilIT.java +++ b/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 data() { - return Arrays.asList("use_gcsutil_v1", "use_gcsutil_v2"); + return Arrays.asList("disable_gcsutil_v2", "use_gcsutil_v2"); } @Parameter public String experiment; diff --git a/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilTest.java b/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilTest.java index 12d89dec705f..b1558322e683 100644 --- a/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilTest.java +++ b/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilTest.java @@ -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; @@ -439,10 +441,42 @@ 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); + } + + /** 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; diff --git a/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilV1Test.java b/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilV1Test.java index 72d554da7528..2295ecdd1cfa 100644 --- a/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilV1Test.java +++ b/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilV1Test.java @@ -105,6 +105,7 @@ import org.apache.beam.sdk.io.fs.MoveOptions.StandardMoveOptions; import org.apache.beam.sdk.metrics.MetricName; import org.apache.beam.sdk.metrics.MetricsEnvironment; +import org.apache.beam.sdk.options.ExperimentalOptions; import org.apache.beam.sdk.options.PipelineOptions; import org.apache.beam.sdk.options.PipelineOptionsFactory; import org.apache.beam.sdk.util.FastNanoClockAndSleeper; @@ -142,8 +143,20 @@ public void tearDown() { MetricsEnvironment.setCurrentContainer(null); } - private static GcsOptions gcsOptionsWithTestCredential() { + /** + * These tests exercise {@link GcsUtilV1}, which is no longer the default, so every {@link + * GcsUtil} they build must opt into it with the {@code disable_gcsutil_v2} experiment. + */ + private static GcsOptions gcsOptionsUsingV1() { GcsOptions pipelineOptions = PipelineOptionsFactory.as(GcsOptions.class); + pipelineOptions + .as(ExperimentalOptions.class) + .setExperiments(Collections.singletonList(GcsUtil.DISABLE_GCSUTIL_V2_EXPERIMENT)); + return pipelineOptions; + } + + private static GcsOptions gcsOptionsWithTestCredential() { + GcsOptions pipelineOptions = gcsOptionsUsingV1(); pipelineOptions.setGcpCredential(new TestCredential()); return pipelineOptions; } @@ -181,7 +194,7 @@ public void testCreationWithExplicitGoogleCloudStorageReadOptions() throws Excep .setFastFailOnNotFoundEnabled(false) .build(); - GcsOptions pipelineOptions = PipelineOptionsFactory.as(GcsOptions.class); + GcsOptions pipelineOptions = gcsOptionsUsingV1(); pipelineOptions.setGoogleCloudStorageReadOptions(readOptions); GcsUtil gcsUtil = pipelineOptions.getGcsUtil(); @@ -1746,7 +1759,7 @@ public void testCreateWithException() throws IOException { private void testWriteMetrics(boolean enabled) throws IOException { // arrange - GcsOptions gcsOptions = PipelineOptionsFactory.create().as(GcsOptions.class); + GcsOptions gcsOptions = gcsOptionsUsingV1(); gcsOptions.setEnableBucketWriteMetricCounter(enabled); gcsOptions.setGcsWriteCounterPrefix("test_counter"); GcsUtilMock gcsUtil = GcsUtilMock.createMockWithMockStorage(gcsOptions, null); @@ -1774,7 +1787,7 @@ private void testWriteMetrics(boolean enabled) throws IOException { private void testReadMetrics(boolean enabled, GoogleCloudStorageReadOptions readOptions) throws IOException { // arrange - GcsOptions gcsOptions = PipelineOptionsFactory.create().as(GcsOptions.class); + GcsOptions gcsOptions = gcsOptionsUsingV1(); gcsOptions.setEnableBucketReadMetricCounter(enabled); gcsOptions.setGcsReadCounterPrefix("test_counter"); if (readOptions != null) { @@ -1833,7 +1846,7 @@ public void testReadMetricsAreNotCollectedWhenNotEnabledOpenWithOptions() throws @Test public void testGcsEndpoint() throws IOException { - GcsOptions pipelineOptions = PipelineOptionsFactory.as(GcsOptions.class); + GcsOptions pipelineOptions = gcsOptionsUsingV1(); pipelineOptions.setGcsEndpoint("http://localhost:4443/storage/v1/"); GcsUtil gcsUtil = pipelineOptions.getGcsUtil(); From 8f6945d94d613be9e834fad51064075b6db501c4 Mon Sep 17 00:00:00 2001 From: Shunping Huang Date: Wed, 21 Oct 2026 14:39:42 -0400 Subject: [PATCH 2/7] Fallback to GcsUtil V1 when use_grpc_for_gcs is set --- .../beam/sdk/extensions/gcp/util/GcsUtil.java | 14 +++++++++++++- .../beam/sdk/extensions/gcp/util/GcsUtilV1.java | 2 +- .../beam/sdk/extensions/gcp/util/GcsUtilTest.java | 12 ++++++++++++ 3 files changed, 26 insertions(+), 2 deletions(-) diff --git a/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtil.java b/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtil.java index e9e34e3cc2b7..ca90b54cadaa 100644 --- a/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtil.java +++ b/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtil.java @@ -140,13 +140,25 @@ public static boolean isWildcard(GcsPath spec) { */ 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); + @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.", - DISABLE_GCSUTIL_V2_EXPERIMENT); + v1Experiment); } else { this.delegateV2 = new GcsUtilV2.GcsUtilFactory().create(options); LOG.info("Using GcsUtilV2 (java-storage) for GCS operations."); diff --git a/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilV1.java b/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilV1.java index 56de6a477aac..bb5570df79c8 100644 --- a/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilV1.java +++ b/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(), diff --git a/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilTest.java b/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilTest.java index b1558322e683..ad30f6f0c491 100644 --- a/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilTest.java +++ b/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilTest.java @@ -473,6 +473,18 @@ public void testFallsBackToV1WithDisableV2Experiment() { .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() { From 3cd193e47986a473de9c7691dd59e5f070af7686 Mon Sep 17 00:00:00 2001 From: Shunping Huang Date: Wed, 21 Oct 2026 14:41:58 -0400 Subject: [PATCH 3/7] Select GcsUtil version explicitly in ParquetIOLT --- .../beam/it/gcp/storage/ParquetIOLT.java | 40 ++++++++++++------- 1 file changed, 25 insertions(+), 15 deletions(-) diff --git a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/storage/ParquetIOLT.java b/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/storage/ParquetIOLT.java index 52b12372dcd6..b8838a6ba246 100644 --- a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/storage/ParquetIOLT.java +++ b/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/storage/ParquetIOLT.java @@ -132,15 +132,17 @@ *

GcsUtil v1 or v2

* *

{@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: * * 

  * -Dconfiguration='{"preset":"f100_s16","useGcsUtilV2":true}'
  *
* - *

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. + *

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. * *

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 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. * - *

Which one a run actually used is logged by {@code GcsUtil} at INFO. + *

{@code GcsUtil} logs the version a run actually used at INFO. */ @JsonProperty public boolean useGcsUtilV2 = false; From 67a0a2d4b4a8e909d16c62e22c4776aab97ae6fd Mon Sep 17 00:00:00 2001 From: Shunping Huang Date: Wed, 21 Oct 2026 14:50:25 -0400 Subject: [PATCH 4/7] Edited CHANGES.md --- CHANGES.md | 1 + 1 file changed, 1 insertion(+) diff --git a/CHANGES.md b/CHANGES.md index 4e2fffb89380..f460f1b0139d 100644 --- a/CHANGES.md +++ b/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 From 897a6fc0d609264dbc3a0eb8af352d3ac591df97 Mon Sep 17 00:00:00 2001 From: Shunping Huang Date: Wed, 21 Oct 2026 15:10:10 -0400 Subject: [PATCH 5/7] Fix GcsUtilIT.testRewriteMultiPart to keep the V1 experiment --- .../org/apache/beam/sdk/extensions/gcp/util/GcsUtilIT.java | 3 +++ 1 file changed, 3 insertions(+) diff --git a/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilIT.java b/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilIT.java index 423c40fe8b4b..adb694677851 100644 --- a/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilIT.java +++ b/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilIT.java @@ -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()); From b7a8e97b9e40d34e4d3b2014d36bfc945e592249 Mon Sep 17 00:00:00 2001 From: Shunping Huang Date: Wed, 21 Oct 2026 15:42:42 -0400 Subject: [PATCH 6/7] Trigger Java PostCommits for GcsUtil V2 default --- .github/trigger_files/beam_PostCommit_Java.json | 2 +- .github/trigger_files/beam_PostCommit_Java_DataflowV1.json | 2 +- .github/trigger_files/beam_PostCommit_Java_DataflowV2.json | 2 +- .../trigger_files/beam_PostCommit_Java_Examples_Dataflow.json | 4 ++++ .../beam_PostCommit_Java_Examples_Dataflow_V2.json | 2 +- .../trigger_files/beam_PostCommit_Java_Examples_Direct.json | 4 ++++ .../trigger_files/beam_PostCommit_Java_Hadoop_Versions.json | 2 +- .../beam_PostCommit_Java_IO_Performance_Tests.json | 2 +- 8 files changed, 14 insertions(+), 6 deletions(-) create mode 100644 .github/trigger_files/beam_PostCommit_Java_Examples_Dataflow.json create mode 100644 .github/trigger_files/beam_PostCommit_Java_Examples_Direct.json diff --git a/.github/trigger_files/beam_PostCommit_Java.json b/.github/trigger_files/beam_PostCommit_Java.json index de320ac8499c..b2a8beaf3775 100644 --- a/.github/trigger_files/beam_PostCommit_Java.json +++ b/.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 } diff --git a/.github/trigger_files/beam_PostCommit_Java_DataflowV1.json b/.github/trigger_files/beam_PostCommit_Java_DataflowV1.json index c403293a8561..dc9b909a926c 100644 --- a/.github/trigger_files/beam_PostCommit_Java_DataflowV1.json +++ b/.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" } diff --git a/.github/trigger_files/beam_PostCommit_Java_DataflowV2.json b/.github/trigger_files/beam_PostCommit_Java_DataflowV2.json index cf92923c2d23..98161d37e42d 100644 --- a/.github/trigger_files/beam_PostCommit_Java_DataflowV2.json +++ b/.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" } diff --git a/.github/trigger_files/beam_PostCommit_Java_Examples_Dataflow.json b/.github/trigger_files/beam_PostCommit_Java_Examples_Dataflow.json new file mode 100644 index 000000000000..e3d6056a5de9 --- /dev/null +++ b/.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 +} diff --git a/.github/trigger_files/beam_PostCommit_Java_Examples_Dataflow_V2.json b/.github/trigger_files/beam_PostCommit_Java_Examples_Dataflow_V2.json index 2bbe49ea6083..5544cd2ef706 100644 --- a/.github/trigger_files/beam_PostCommit_Java_Examples_Dataflow_V2.json +++ b/.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" } diff --git a/.github/trigger_files/beam_PostCommit_Java_Examples_Direct.json b/.github/trigger_files/beam_PostCommit_Java_Examples_Direct.json new file mode 100644 index 000000000000..e3d6056a5de9 --- /dev/null +++ b/.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 +} diff --git a/.github/trigger_files/beam_PostCommit_Java_Hadoop_Versions.json b/.github/trigger_files/beam_PostCommit_Java_Hadoop_Versions.json index f1ba03a243ee..455144f02a35 100644 --- a/.github/trigger_files/beam_PostCommit_Java_Hadoop_Versions.json +++ b/.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 } diff --git a/.github/trigger_files/beam_PostCommit_Java_IO_Performance_Tests.json b/.github/trigger_files/beam_PostCommit_Java_IO_Performance_Tests.json index 12baa399f316..b2fec41d7e1b 100644 --- a/.github/trigger_files/beam_PostCommit_Java_IO_Performance_Tests.json +++ b/.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, } From b4eb5477a9e6d2aaccadf332ef57506e9a71df33 Mon Sep 17 00:00:00 2001 From: Shunping Huang Date: Wed, 21 Oct 2026 17:09:02 -0400 Subject: [PATCH 7/7] Preserve kmsKeyName in GcsUtil V2 getObject. Fix test. --- .../java/org/apache/beam/sdk/extensions/gcp/util/GcsUtil.java | 3 ++- .../org/apache/beam/sdk/extensions/gcp/util/GcsUtilTest.java | 2 ++ 2 files changed, 4 insertions(+), 1 deletion(-) diff --git a/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtil.java b/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtil.java index ca90b54cadaa..25ca962711d0 100644 --- a/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtil.java +++ b/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtil.java @@ -658,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)); diff --git a/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilTest.java b/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilTest.java index ad30f6f0c491..2501313304e1 100644 --- a/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilTest.java +++ b/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilTest.java @@ -335,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()) @@ -353,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);