GCP: add pluggable GCS token credential provider

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

@@ -67,6 +67,28 @@ public class GCPProperties implements Serializable {
   public static final String GCS_OAUTH2_REFRESH_CREDENTIALS_ENABLED =
       "gcs.oauth2.refresh-credentials-enabled";
 
+  /**
+   * Class name of a custom {@link org.apache.iceberg.gcp.gcs.GcsTokenCredentialProvider}
+   * implementation. When set, {@code PrefixedStorage} uses this provider to obtain {@link
+   * com.google.auth.oauth2.GoogleCredentials} instead of reading a static {@link #GCS_OAUTH2_TOKEN}
+   * property. The provider class must have a no-arg constructor.
+   *
+   * <p><b>Precedence:</b> This property is ignored if {@link #GCS_OAUTH2_TOKEN} or {@link
+   * #GCS_IMPERSONATE_SERVICE_ACCOUNT} is also set. The vended credential path sets both {@link
+   * #GCS_OAUTH2_TOKEN} (per-prefix token) and this provider (refresh source), so the token takes
+   * precedence. Impersonation similarly takes precedence to maintain deterministic credential
+   * selection.
+   */
+  public static final String GCS_TOKEN_CREDENTIAL_PROVIDER = "gcs.token-credential-provider";
+
+  /**
+   * Property prefix for initializing a custom {@link
+   * org.apache.iceberg.gcp.gcs.GcsTokenCredentialProvider}. All properties under this prefix are
+   * extracted (without the prefix) and passed to {@link
+   * org.apache.iceberg.gcp.gcs.GcsTokenCredentialProvider#initialize(Map)}.
+   */
+  public static final String GCS_TOKEN_PROVIDER_PREFIX = "gcs.token-credential-provider.";
+
   /** 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";
 
@@ -98,6 +120,7 @@ public class GCPProperties implements Serializable {
   private String gcsOauth2RefreshCredentialsEndpoint;
   private boolean gcsOauth2RefreshCredentialsEnabled;
   private boolean gcsAnalyticsCoreEnabled;
+  private String gcsTokenCredentialProvider;
 
   private String gcsImpersonateServiceAccount;
   private int gcsImpersonateLifetimeSeconds;
@@ -167,6 +190,8 @@ public GCPProperties(Map<String, String> properties) {
           new Date(Long.parseLong(properties.get(GCS_OAUTH2_TOKEN_EXPIRES_AT)));
     }
 
+    gcsTokenCredentialProvider = properties.get(GCS_TOKEN_CREDENTIAL_PROVIDER);
+
     gcsOauth2RefreshCredentialsEndpoint =
         RESTUtil.resolveEndpoint(
             properties.get(CatalogProperties.URI),
@@ -179,6 +204,11 @@ public GCPProperties(Map<String, String> properties) {
         "Invalid auth settings: must not configure %s and %s",
         GCS_NO_AUTH,
         GCS_OAUTH2_TOKEN);
+    Preconditions.checkState(
+        !(gcsTokenCredentialProvider != null && gcsNoAuth),
+        "Invalid auth settings: must not configure %s and %s",
+        GCS_NO_AUTH,
+        GCS_TOKEN_CREDENTIAL_PROVIDER);
 
     gcsDeleteBatchSize =
         PropertyUtil.propertyAsInt(
@@ -235,6 +265,10 @@ public Optional<String> oauth2Token() {
     return Optional.ofNullable(gcsOAuth2Token);
   }
 
+  public Optional<String> tokenCredentialProvider() {
+    return Optional.ofNullable(gcsTokenCredentialProvider);
+  }
+
   public boolean noAuth() {
     return gcsNoAuth;
   }

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

@@ -0,0 +1,39 @@
+/*
+ * 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.gcs;
+
+import com.google.auth.oauth2.GoogleCredentials;
+import java.util.Map;
+
+public interface GcsTokenCredentialProvider {
+
+  /**
+   * Returns a GoogleCredentials instance for authenticating GCS requests.
+   *
+   * @return a GoogleCredentials instance, must not be null
+   */
+  GoogleCredentials credential();
+
+  /**
+   * Initialize GCS credential provider from provider properties.
+   *
+   * @param properties credential provider properties
+   */
+  void initialize(Map<String, String> properties);
+}

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

@@ -0,0 +1,102 @@
+/*
+ * 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.gcs;
+
+import com.google.auth.oauth2.GoogleCredentials;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.Map;
+import org.apache.iceberg.common.DynConstructors;
+import org.apache.iceberg.gcp.GCPProperties;
+import org.apache.iceberg.relocated.com.google.common.base.Strings;
+import org.apache.iceberg.util.PropertyUtil;
+
+public class GcsTokenCredentialProviders {
+
+  private static final DefaultGcsTokenCredentialProvider DEFAULT_PROVIDER =
+      new DefaultGcsTokenCredentialProvider();
+
+  private GcsTokenCredentialProviders() {}
+
+  public static GcsTokenCredentialProvider defaultFactory() {
+    return DEFAULT_PROVIDER;
+  }
+
+  public static GcsTokenCredentialProvider from(Map<String, String> properties) {
+    String providerImpl =
+        PropertyUtil.propertyAsString(
+            properties, GCPProperties.GCS_TOKEN_CREDENTIAL_PROVIDER, null);
+    Map<String, String> credentialProviderProperties =
+        PropertyUtil.propertiesWithPrefix(properties, GCPProperties.GCS_TOKEN_PROVIDER_PREFIX);
+    return loadCredentialProvider(providerImpl, credentialProviderProperties);
+  }
+
+  private static GcsTokenCredentialProvider loadCredentialProvider(
+      String impl, Map<String, String> properties) {
+    if (Strings.isNullOrEmpty(impl)) {
+      GcsTokenCredentialProvider provider = defaultFactory();
+      provider.initialize(properties);
+      return provider;
+    }
+
+    DynConstructors.Ctor<GcsTokenCredentialProvider> ctor;
+    try {
+      ctor =
+          DynConstructors.builder(GcsTokenCredentialProvider.class)
+              .loader(GcsTokenCredentialProviders.class.getClassLoader())
+              .hiddenImpl(impl)
+              .buildChecked();
+    } catch (NoSuchMethodException e) {
+      throw new IllegalArgumentException(
+          String.format(
+              "Cannot initialize GcsTokenCredentialProvider, cannot load class or missing no-arg constructor: %s",
+              impl),
+          e);
+    }
+
+    GcsTokenCredentialProvider provider;
+    try {
+      provider = ctor.newInstance();
+    } catch (ClassCastException e) {
+      throw new IllegalArgumentException(
+          String.format(
+              "Cannot initialize GcsTokenCredentialProvider, %s does not implement GcsTokenCredentialProvider.",
+              impl),
+          e);
+    }
+
+    provider.initialize(properties);
+    return provider;
+  }
+
+  static class DefaultGcsTokenCredentialProvider implements GcsTokenCredentialProvider {
+
+    @Override
+    public GoogleCredentials credential() {
+      try {
+        return GoogleCredentials.getApplicationDefault();
+      } catch (IOException e) {
+        throw new UncheckedIOException("Failed to get application default GCS credentials", e);
+      }
+    }
+
+    @Override
+    public void initialize(Map<String, String> properties) {}
+  }
+}

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

@@ -36,8 +36,11 @@
 import org.apache.iceberg.relocated.com.google.common.base.Strings;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
 import org.apache.iceberg.util.SerializableSupplier;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 class PrefixedStorage implements AutoCloseable {
+  private static final Logger LOG = LoggerFactory.getLogger(PrefixedStorage.class);
   private static final String GCS_FILE_IO_USER_AGENT = "gcsfileio/" + EnvironmentContext.get();
   private final String storagePrefix;
   private final GCPProperties gcpProperties;
@@ -157,7 +160,24 @@ private Credentials credentials(GCPProperties properties) {
       // Explicitly allow "no credentials" for testing purposes
       return NoCredentials.getInstance();
     } else if (properties.impersonateServiceAccount().isPresent()) {
+      if (gcpProperties.tokenCredentialProvider().isPresent()) {
+        LOG.warn(
+            "Both {} and {} are set; {} takes precedence and the provider is ignored",
+            GCPProperties.GCS_IMPERSONATE_SERVICE_ACCOUNT,
+            GCPProperties.GCS_TOKEN_CREDENTIAL_PROVIDER,
+            GCPProperties.GCS_IMPERSONATE_SERVICE_ACCOUNT);
+      }
       return buildImpersonatedCredentials(properties);
+    } else if (properties.tokenCredentialProvider().isPresent()) {
+      // A custom provider yields a self-refreshing GoogleCredentials (e.g. built from a
+      // caller-supplied source credential), addressing static-token expiry for non-vended setups.
+      GoogleCredentials credentials =
+          GcsTokenCredentialProviders.from(properties.properties()).credential();
+      Preconditions.checkState(
+          credentials != null,
+          "Provider %s returned null credentials",
+          gcpProperties.tokenCredentialProvider().get());
+      return credentials;
     } else {
       return null;
     }

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

@@ -22,6 +22,7 @@
 import static org.apache.iceberg.gcp.GCPProperties.GCS_OAUTH2_REFRESH_CREDENTIALS_ENABLED;
 import static org.apache.iceberg.gcp.GCPProperties.GCS_OAUTH2_REFRESH_CREDENTIALS_ENDPOINT;
 import static org.apache.iceberg.gcp.GCPProperties.GCS_OAUTH2_TOKEN;
+import static org.apache.iceberg.gcp.GCPProperties.GCS_TOKEN_CREDENTIAL_PROVIDER;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatIllegalStateException;
 
@@ -50,6 +51,51 @@ public void testOAuthWithNoAuth() {
     assertThat(gcpProperties.oauth2Token()).isNotPresent();
   }
 
+  @Test
+  public void testTokenCredentialProviderWithNoAuth() {
+    assertThatIllegalStateException()
+        .isThrownBy(
+            () ->
+                new GCPProperties(
+                    ImmutableMap.of(
+                        GCS_TOKEN_CREDENTIAL_PROVIDER,
+                        "org.example.Provider",
+                        GCS_NO_AUTH,
+                        "true")))
+        .withMessage(
+            String.format(
+                "Invalid auth settings: must not configure %s and %s",
+                GCS_NO_AUTH, GCS_TOKEN_CREDENTIAL_PROVIDER));
+  }
+
+  @Test
+  public void testTokenCredentialProviderWithOAuth2Token() {
+    // Provider and oauth2 token may coexist (the vended case) - construction must not throw;
+    // PrefixedStorage resolves precedence.
+    GCPProperties gcpProperties =
+        new GCPProperties(
+            ImmutableMap.of(
+                GCS_TOKEN_CREDENTIAL_PROVIDER, "org.example.Provider", GCS_OAUTH2_TOKEN, "oauth"));
+    assertThat(gcpProperties.tokenCredentialProvider())
+        .isPresent()
+        .get()
+        .isEqualTo("org.example.Provider");
+    assertThat(gcpProperties.oauth2Token()).isPresent().get().isEqualTo("oauth");
+    assertThat(gcpProperties.noAuth()).isFalse();
+  }
+
+  @Test
+  public void testTokenCredentialProviderSet() {
+    GCPProperties gcpProperties =
+        new GCPProperties(ImmutableMap.of(GCS_TOKEN_CREDENTIAL_PROVIDER, "org.example.Provider"));
+    assertThat(gcpProperties.tokenCredentialProvider())
+        .isPresent()
+        .get()
+        .isEqualTo("org.example.Provider");
+    assertThat(gcpProperties.oauth2Token()).isNotPresent();
+    assertThat(gcpProperties.noAuth()).isFalse();
+  }
+
   @Test
   public void refreshCredentialsEndpointSet() {
     GCPProperties gcpProperties =