GCS: Allow no-auth for testing purposes (#9061)

Although there is no "official" Google Cloud Storage emulator available yet, there is [one available](https://github.com/oittaa/gcp-storage-emulator) that allows at least some basic testing. To use an emulator, the client needs to be configured to use no authentication, otherwise it will fallback to "automatic credential detection".

gcp/src/main/java/org/apache/iceberg/gcp/GCPProperties.java

@@ -22,6 +22,7 @@
 import java.util.Date;
 import java.util.Map;
 import java.util.Optional;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.util.PropertyUtil;
 
 public class GCPProperties implements Serializable {
@@ -40,6 +41,8 @@ public class GCPProperties implements Serializable {
 
   public static final String GCS_OAUTH2_TOKEN = "gcs.oauth2.token";
   public static final String GCS_OAUTH2_TOKEN_EXPIRES_AT = "gcs.oauth2.token-expires-at";
+  // Boolean to explicitly configure "no authentication" for testing purposes using a GCS emulator
+  public static final String GCS_NO_AUTH = "gcs.no-auth";
 
   /** Configure the batch size used when deleting multiple files from a given GCS bucket */
   public static final String GCS_DELETE_BATCH_SIZE = "gcs.delete.batch-size";
@@ -60,6 +63,7 @@ public class GCPProperties implements Serializable {
   private Integer gcsChannelReadChunkSize;
   private Integer gcsChannelWriteChunkSize;
 
+  private boolean gcsNoAuth;
   private String gcsOAuth2Token;
   private Date gcsOAuth2TokenExpiresAt;
 
@@ -90,6 +94,12 @@ public GCPProperties(Map<String, String> properties) {
       gcsOAuth2TokenExpiresAt =
           new Date(Long.parseLong(properties.get(GCS_OAUTH2_TOKEN_EXPIRES_AT)));
     }
+    gcsNoAuth = Boolean.parseBoolean(properties.getOrDefault(GCS_NO_AUTH, "false"));
+    Preconditions.checkState(
+        !(gcsOAuth2Token != null && gcsNoAuth),
+        "Invalid auth settings: must not configure %s and %s",
+        GCS_NO_AUTH,
+        GCS_OAUTH2_TOKEN);
 
     gcsDeleteBatchSize =
         PropertyUtil.propertyAsInt(
@@ -132,6 +142,10 @@ public Optional<String> oauth2Token() {
     return Optional.ofNullable(gcsOAuth2Token);
   }
 
+  public boolean noAuth() {
+    return gcsNoAuth;
+  }
+
   public Optional<Date> oauth2TokenExpiresAt() {
     return Optional.ofNullable(gcsOAuth2TokenExpiresAt);
   }

gcp/src/main/java/org/apache/iceberg/gcp/gcs/GCSFileIO.java

@@ -20,6 +20,7 @@
 
 import com.google.auth.oauth2.AccessToken;
 import com.google.auth.oauth2.OAuth2Credentials;
+import com.google.cloud.NoCredentials;
 import com.google.cloud.storage.Blob;
 import com.google.cloud.storage.BlobId;
 import com.google.cloud.storage.Storage;
@@ -141,10 +142,18 @@ public void initialize(Map<String, String> props) {
           gcpProperties.clientLibToken().ifPresent(builder::setClientLibToken);
           gcpProperties.serviceHost().ifPresent(builder::setHost);
 
+          // Google Cloud APIs default to automatically detect the credentials to use, which is
+          // in most cases the convenient way, especially in GCP.
+          // See javadoc of com.google.auth.oauth2.GoogleCredentials.getApplicationDefault().
+          if (gcpProperties.noAuth()) {
+            // Explicitly allow "no credentials" for testing purposes.
+            builder.setCredentials(NoCredentials.getInstance());
+          }
           gcpProperties
               .oauth2Token()
               .ifPresent(
                   token -> {
+                    // Explicitly configure an OAuth token.
                     AccessToken accessToken =
                         new AccessToken(token, gcpProperties.oauth2TokenExpiresAt().orElse(null));
                     builder.setCredentials(OAuth2Credentials.create(accessToken));

gcp/src/test/java/org/apache/iceberg/gcp/GCPPropertiesTest.java

@@ -0,0 +1,50 @@
+/*
+ * 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.iceberg.gcp;
+
+import static org.apache.iceberg.gcp.GCPProperties.GCS_NO_AUTH;
+import static org.apache.iceberg.gcp.GCPProperties.GCS_OAUTH2_TOKEN;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.assertj.core.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class GCPPropertiesTest {
+
+  @Test
+  public void testOAuthWithNoAuth() {
+    Assertions.assertThatIllegalStateException()
+        .isThrownBy(
+            () ->
+                new GCPProperties(ImmutableMap.of(GCS_OAUTH2_TOKEN, "oauth", GCS_NO_AUTH, "true")))
+        .withMessage(
+            String.format(
+                "Invalid auth settings: must not configure %s and %s",
+                GCS_NO_AUTH, GCS_OAUTH2_TOKEN));
+
+    GCPProperties gcpProperties =
+        new GCPProperties(ImmutableMap.of(GCS_OAUTH2_TOKEN, "oauth", GCS_NO_AUTH, "false"));
+    assertThat(gcpProperties.noAuth()).isFalse();
+    assertThat(gcpProperties.oauth2Token()).get().isEqualTo("oauth");
+    gcpProperties = new GCPProperties(ImmutableMap.of(GCS_NO_AUTH, "true"));
+    assertThat(gcpProperties.noAuth()).isTrue();
+    assertThat(gcpProperties.oauth2Token()).isNotPresent();
+  }
+}