Fill gaps between GcsUtilV2 and GcsUtilV1

buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy

@@ -768,6 +768,7 @@ class BeamModulePlugin implements Plugin<Project> {
         google_cloud_bigtable_emulator              : "com.google.cloud:google-cloud-bigtable-emulator", // google_cloud_platform_libraries_bom sets version
         google_cloud_core                           : "com.google.cloud:google-cloud-core", // google_cloud_platform_libraries_bom sets version
         google_cloud_core_grpc                      : "com.google.cloud:google-cloud-core-grpc", // google_cloud_platform_libraries_bom sets version
+        google_cloud_core_http                      : "com.google.cloud:google-cloud-core-http", // google_cloud_platform_libraries_bom sets version
         google_cloud_datacatalog_v1beta1            : "com.google.cloud:google-cloud-datacatalog", // google_cloud_platform_libraries_bom sets version
         google_cloud_dataflow_java_proto_library_all: "com.google.cloud.dataflow:google-cloud-dataflow-java-proto-library-all:0.5.160304",
         google_cloud_datastore_v1_proto_client      : "com.google.cloud.datastore:datastore-v1-proto-client:3.4.0",   // [bomupgrader] sets version

sdks/java/extensions/google-cloud-platform-core/build.gradle

@@ -44,6 +44,7 @@ dependencies {
   implementation library.java.google_auth_library_oauth2_http
   implementation library.java.google_api_client
   implementation library.java.google_cloud_core
+  implementation library.java.google_cloud_core_http
   implementation library.java.google_cloud_storage
   implementation library.java.bigdataoss_gcsio
   implementation library.java.bigdataoss_util

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

@@ -47,8 +47,12 @@
 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.Sets;
 import org.checkerframework.checker.nullness.qual.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class GcsUtil {
+  private static final Logger LOG = LoggerFactory.getLogger(GcsUtil.class);
+
   /**
    * Namespace for every GCS metric. The namespace is dropped when Dataflow exports counters to
    * Cloud Monitoring, so the layer is carried by the metric name instead: {@code gcs_http_*} for
@@ -119,8 +123,12 @@ public static boolean isWildcard(GcsPath spec) {
     this.delegate = new GcsUtilV1.GcsUtilFactory().create(options);
     if (ExperimentalOptions.hasExperiment(options, "use_gcsutil_v2")) {
       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.");
     }
   }
 
@@ -297,7 +305,7 @@ public CreateOptions build() {
 
   public WritableByteChannel create(GcsPath path, CreateOptions options) throws IOException {
     if (delegateV2 != null) {
-      delegateV2.create(path, options.delegate);
+      return delegateV2.create(path, options.delegate);
     }
     return delegate.create(path, options.delegate);
   }