This is an automated email from the ASF dual-hosted git repository.
shunping pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 94b84b6adcb [Java] Use GcsUtil V2 by default (#40713)
94b84b6adcb is described below
commit 94b84b6adcb106b95afb50123645ae52a8d81d7a
Author: Shunping Huang <[email protected]>
AuthorDate: Thu Oct 22 17:16:33 2026 -0400
[Java] Use GcsUtil V2 by default (#40713)
* 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.
* Fallback to GcsUtil V1 when use_grpc_for_gcs is set
* Select GcsUtil version explicitly in ParquetIOLT
* Edited CHANGES.md
* Fix GcsUtilIT.testRewriteMultiPart to keep the V1 experiment
* Trigger Java PostCommits for GcsUtil V2 default
* Preserve kmsKeyName in GcsUtil V2 getObject. Fix test.
---
.github/trigger_files/beam_PostCommit_Java.json | 2 +-
.../beam_PostCommit_Java_DataflowV1.json | 2 +-
.../beam_PostCommit_Java_DataflowV2.json | 2 +-
...=> beam_PostCommit_Java_Examples_Dataflow.json} | 2 +-
.../beam_PostCommit_Java_Examples_Dataflow_V2.json | 2 +-
...n => beam_PostCommit_Java_Examples_Direct.json} | 2 +-
.../beam_PostCommit_Java_Hadoop_Versions.json | 2 +-
.../beam_PostCommit_Java_IO_Performance_Tests.json | 2 +-
CHANGES.md | 1 +
.../apache/beam/it/gcp/storage/ParquetIOLT.java | 40 ++++++++++------
.../sdk/extensions/gcp/options/GcsOptions.java | 19 ++++----
.../beam/sdk/extensions/gcp/util/GcsUtil.java | 40 +++++++++++++---
.../beam/sdk/extensions/gcp/util/GcsUtilV1.java | 2 +-
.../beam/sdk/extensions/gcp/util/GcsUtilIT.java | 5 +-
.../beam/sdk/extensions/gcp/util/GcsUtilTest.java | 54 ++++++++++++++++++++--
.../sdk/extensions/gcp/util/GcsUtilV1Test.java | 23 +++++++--
16 files changed, 151 insertions(+), 49 deletions(-)
diff --git a/.github/trigger_files/beam_PostCommit_Java.json
b/.github/trigger_files/beam_PostCommit_Java.json
index de320ac8499..b2a8beaf377 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 c403293a856..dc9b909a926 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 cf92923c2d2..98161d37e42 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_Hadoop_Versions.json
b/.github/trigger_files/beam_PostCommit_Java_Examples_Dataflow.json
similarity index 81%
copy from .github/trigger_files/beam_PostCommit_Java_Hadoop_Versions.json
copy to .github/trigger_files/beam_PostCommit_Java_Examples_Dataflow.json
index f1ba03a243e..e3d6056a5de 100644
--- a/.github/trigger_files/beam_PostCommit_Java_Hadoop_Versions.json
+++ b/.github/trigger_files/beam_PostCommit_Java_Examples_Dataflow.json
@@ -1,4 +1,4 @@
{
"comment": "Modify this file in a trivial way to cause this test suite to
run",
- "modification": 5
+ "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 2bbe49ea608..5544cd2ef70 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_Hadoop_Versions.json
b/.github/trigger_files/beam_PostCommit_Java_Examples_Direct.json
similarity index 81%
copy from .github/trigger_files/beam_PostCommit_Java_Hadoop_Versions.json
copy to .github/trigger_files/beam_PostCommit_Java_Examples_Direct.json
index f1ba03a243e..e3d6056a5de 100644
--- a/.github/trigger_files/beam_PostCommit_Java_Hadoop_Versions.json
+++ b/.github/trigger_files/beam_PostCommit_Java_Examples_Direct.json
@@ -1,4 +1,4 @@
{
"comment": "Modify this file in a trivial way to cause this test suite to
run",
- "modification": 5
+ "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 f1ba03a243e..455144f02a3 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 12baa399f31..b2fec41d7e1 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,
}
diff --git a/CHANGES.md b/CHANGES.md
index 1f57af6cdb7..77f03f3da76 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
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 52b12372dcd..b8838a6ba24 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 @@ import org.junit.runners.MethodSorters;
* <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 @@ public final class ParquetIOLT extends GcsIOLoadTestBase {
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 @@ public final class ParquetIOLT extends GcsIOLoadTestBase {
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 @@ public final class ParquetIOLT extends GcsIOLoadTestBase {
@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;
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 b5246a87d2b..88ebf73dfab 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 interface GcsOptions extends
ApplicationNameOptions, GcpOptions, Pipeline
/**
* 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 interface GcsOptions extends
ApplicationNameOptions, GcpOptions, Pipeline
* 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);
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 382ac6856bf..25ca962711d 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,41 @@ public class GcsUtil {
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 @@ public class GcsUtil {
.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/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 56de6a477aa..bb5570df79c 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 @@ class GcsUtilV1 {
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/GcsUtilIT.java
b/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilIT.java
index a6ae12024a1..adb69467785 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<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 class GcsUtilIT {
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());
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 12d89dec705..2501313304e 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.nio.channels.SeekableByteChannel;
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.options.GcsOptions;
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 class GcsUtilTest {
}
// 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 class GcsUtilTest {
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 class GcsUtilTest {
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 class GcsUtilTest {
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;
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 72d554da752..2295ecdd1cf 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.extensions.gcp.util.gcsfs.GcsPath;
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 class GcsUtilV1Test {
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 class GcsUtilV1Test {
.setFastFailOnNotFoundEnabled(false)
.build();
- GcsOptions pipelineOptions = PipelineOptionsFactory.as(GcsOptions.class);
+ GcsOptions pipelineOptions = gcsOptionsUsingV1();
pipelineOptions.setGoogleCloudStorageReadOptions(readOptions);
GcsUtil gcsUtil = pipelineOptions.getGcsUtil();
@@ -1746,7 +1759,7 @@ public class GcsUtilV1Test {
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 @@ public class GcsUtilV1Test {
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 class GcsUtilV1Test {
@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();
