GCS: Allow no-auth for testing purposes

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

Author: snazy

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

@@ -18,13 +18,15 @@
  */
 package org.apache.iceberg.gcp;
 
+import com.google.api.client.util.Preconditions;
 import java.io.Serializable;
 import java.util.Date;
 import java.util.Map;
 import java.util.Optional;
 import org.apache.iceberg.util.PropertyUtil;
 
 public class GCPProperties implements Serializable {
+
   // Service Options
   public static final String GCS_PROJECT_ID = "gcs.project-id";
   public static final String GCS_CLIENT_LIB_TOKEN = "gcs.client-lib-token";
@@ -40,6 +42,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 +64,7 @@ public class GCPProperties implements Serializable {
   private Integer gcsChannelReadChunkSize;
   private Integer gcsChannelWriteChunkSize;
 
+  private boolean gcsNoAuth;
   private String gcsOAuth2Token;
   private Date gcsOAuth2TokenExpiresAt;
 
@@ -90,6 +95,10 @@ 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),
+        "Must not configure " + GCS_NO_AUTH + " and " + GCS_OAUTH2_TOKEN);
 
     gcsDeleteBatchSize =
         PropertyUtil.propertyAsInt(
@@ -132,6 +141,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;
@@ -55,6 +56,7 @@
  * Overview</a>
  */
 public class GCSFileIO implements DelegateFileIO {
+
   private static final Logger LOG = LoggerFactory.getLogger(GCSFileIO.class);
   private static final String DEFAULT_METRICS_IMPL =
       "org.apache.iceberg.hadoop.HadoopMetricsContext";
@@ -141,10 +143,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));