From 310607a5d1eaa9d15bb223d12b949cf186532a6b Mon Sep 17 00:00:00 2001
From: Bryan Keller <bryanck@gmail.com>
Date: Sat, 15 Jul 2023 06:12:35 -0700
Subject: [PATCH] GCP: Add properties for OAtuh2 and update library

---
 .../apache/iceberg/io/ResolvingFileIO.java    |  4 +++-
 .../org/apache/iceberg/gcp/GCPProperties.java | 21 +++++++++++++++++++
 .../org/apache/iceberg/gcp/gcs/GCSFileIO.java | 11 ++++++++++
 versions.props                                |  2 +-
 4 files changed, 36 insertions(+), 2 deletions(-)

diff --git a/core/src/main/java/org/apache/iceberg/io/ResolvingFileIO.java b/core/src/main/java/org/apache/iceberg/io/ResolvingFileIO.java
index f7369224ee29..28d07bcf249e 100644
--- a/core/src/main/java/org/apache/iceberg/io/ResolvingFileIO.java
+++ b/core/src/main/java/org/apache/iceberg/io/ResolvingFileIO.java
@@ -42,11 +42,13 @@ public class ResolvingFileIO implements FileIO, HadoopConfigurable {
   private static final Logger LOG = LoggerFactory.getLogger(ResolvingFileIO.class);
   private static final String FALLBACK_IMPL = "org.apache.iceberg.hadoop.HadoopFileIO";
   private static final String S3_FILE_IO_IMPL = "org.apache.iceberg.aws.s3.S3FileIO";
+  private static final String GCS_FILE_IO_IMPL = "org.apache.iceberg.gcp.gcs.GCSFileIO";
   private static final Map<String, String> SCHEME_TO_FILE_IO =
       ImmutableMap.of(
           "s3", S3_FILE_IO_IMPL,
           "s3a", S3_FILE_IO_IMPL,
-          "s3n", S3_FILE_IO_IMPL);
+          "s3n", S3_FILE_IO_IMPL,
+          "gs", GCS_FILE_IO_IMPL);
 
   private final Map<String, FileIO> ioInstances = Maps.newHashMap();
   private final AtomicBoolean isClosed = new AtomicBoolean(false);
diff --git a/gcp/src/main/java/org/apache/iceberg/gcp/GCPProperties.java b/gcp/src/main/java/org/apache/iceberg/gcp/GCPProperties.java
index 457a76313d15..109a4d21ed3c 100644
--- a/gcp/src/main/java/org/apache/iceberg/gcp/GCPProperties.java
+++ b/gcp/src/main/java/org/apache/iceberg/gcp/GCPProperties.java
@@ -19,6 +19,7 @@
 package org.apache.iceberg.gcp;
 
 import java.io.Serializable;
+import java.util.Date;
 import java.util.Map;
 import java.util.Optional;
 
@@ -36,6 +37,9 @@ public class GCPProperties implements Serializable {
   public static final String GCS_CHANNEL_READ_CHUNK_SIZE = "gcs.channel.read.chunk-size-bytes";
   public static final String GCS_CHANNEL_WRITE_CHUNK_SIZE = "gcs.channel.write.chunk-size-bytes";
 
+  public static final String GCS_OAUTH2_TOKEN = "gcs.oauth2.token";
+  public static final String GCS_OAUTH2_TOKEN_EXPIRES_AT = "gcs.oauth2.token-expires-at";
+
   private String projectId;
   private String clientLibToken;
   private String serviceHost;
@@ -47,6 +51,9 @@ public class GCPProperties implements Serializable {
   private Integer gcsChannelReadChunkSize;
   private Integer gcsChannelWriteChunkSize;
 
+  private String gcsOAuth2Token;
+  private Date gcsOAuth2TokenExpiresAt;
+
   public GCPProperties() {}
 
   public GCPProperties(Map<String, String> properties) {
@@ -65,6 +72,12 @@ public GCPProperties(Map<String, String> properties) {
     if (properties.containsKey(GCS_CHANNEL_WRITE_CHUNK_SIZE)) {
       gcsChannelWriteChunkSize = Integer.parseInt(properties.get(GCS_CHANNEL_WRITE_CHUNK_SIZE));
     }
+
+    gcsOAuth2Token = properties.get(GCS_OAUTH2_TOKEN);
+    if (properties.containsKey(GCS_OAUTH2_TOKEN_EXPIRES_AT)) {
+      gcsOAuth2TokenExpiresAt =
+          new Date(Long.parseLong(properties.get(GCS_OAUTH2_TOKEN_EXPIRES_AT)));
+    }
   }
 
   public Optional<Integer> channelReadChunkSize() {
@@ -98,4 +111,12 @@ public Optional<String> serviceHost() {
   public Optional<String> userProject() {
     return Optional.ofNullable(gcsUserProject);
   }
+
+  public Optional<String> oauth2Token() {
+    return Optional.ofNullable(gcsOAuth2Token);
+  }
+
+  public Optional<Date> oauth2TokenExpiresAt() {
+    return Optional.ofNullable(gcsOAuth2TokenExpiresAt);
+  }
 }
diff --git a/gcp/src/main/java/org/apache/iceberg/gcp/gcs/GCSFileIO.java b/gcp/src/main/java/org/apache/iceberg/gcp/gcs/GCSFileIO.java
index 390896b9923d..54af44e43da8 100644
--- a/gcp/src/main/java/org/apache/iceberg/gcp/gcs/GCSFileIO.java
+++ b/gcp/src/main/java/org/apache/iceberg/gcp/gcs/GCSFileIO.java
@@ -18,6 +18,8 @@
  */
 package org.apache.iceberg.gcp.gcs;
 
+import com.google.auth.oauth2.AccessToken;
+import com.google.auth.oauth2.OAuth2Credentials;
 import com.google.cloud.storage.BlobId;
 import com.google.cloud.storage.Storage;
 import com.google.cloud.storage.StorageOptions;
@@ -133,6 +135,15 @@ public void initialize(Map<String, String> props) {
           gcpProperties.clientLibToken().ifPresent(builder::setClientLibToken);
           gcpProperties.serviceHost().ifPresent(builder::setHost);
 
+          gcpProperties
+              .oauth2Token()
+              .ifPresent(
+                  token -> {
+                    AccessToken accessToken =
+                        new AccessToken(token, gcpProperties.oauth2TokenExpiresAt().orElse(null));
+                    builder.setCredentials(OAuth2Credentials.create(accessToken));
+                  });
+
           // Report Hadoop metrics if Hadoop is available
           try {
             DynConstructors.Ctor<MetricsContext> ctor =
diff --git a/versions.props b/versions.props
index 020f1d718eb5..07597d4ed90e 100644
--- a/versions.props
+++ b/versions.props
@@ -23,7 +23,7 @@ javax.activation:activation = 1.1.1
 org.glassfish.jaxb:jaxb-runtime = 2.3.3
 software.amazon.awssdk:* = 2.20.18
 org.projectnessie.nessie:* = 0.63.0
-com.google.cloud:libraries-bom = 24.1.0
+com.google.cloud:libraries-bom = 26.18.0
 org.scala-lang.modules:scala-collection-compat_2.12 = 2.11.0
 org.scala-lang.modules:scala-collection-compat_2.13 = 2.11.0
 com.emc.ecs:object-client-bundle = 3.3.2