feat: Allows usage of Iceberg's GoogleAuthManager rather than just OAuth

polaris-core/src/main/java/org/apache/polaris/core/config/FeatureConfiguration.java

@@ -373,6 +373,7 @@ public static void enforceFeatureEnabledOrThrow(
                   List.of(
                       AuthenticationParameters.AuthenticationTypeEnum.OAUTH.name(),
                       AuthenticationParameters.AuthenticationTypeEnum.BEARER.name(),
+                      AuthenticationParameters.AuthenticationTypeEnum.GCP.name(),
                       AuthenticationParameters.AuthenticationTypeEnum.SIGV4.name()))
               .buildFeatureConfiguration();
 

polaris-core/src/main/java/org/apache/polaris/core/connection/AuthenticationParametersDpo.java

@@ -43,6 +43,7 @@
   @JsonSubTypes.Type(value = BearerAuthenticationParametersDpo.class, name = "2"),
   @JsonSubTypes.Type(value = ImplicitAuthenticationParametersDpo.class, name = "3"),
   @JsonSubTypes.Type(value = SigV4AuthenticationParametersDpo.class, name = "4"),
+  @JsonSubTypes.Type(value = GcpAuthenticationParametersDpo.class, name = "5"),
 })
 public abstract class AuthenticationParametersDpo implements IcebergCatalogPropertiesProvider {
 
@@ -73,6 +74,9 @@ public static AuthenticationParametersDpo fromAuthenticationParametersModelWithS
       Map<String, SecretReference> secretReferences) {
     final AuthenticationParametersDpo config;
     switch (authenticationParameters.getAuthenticationType()) {
+      case GCP:
+        config = new GcpAuthenticationParametersDpo();
+        break;
       case OAUTH:
         OAuthClientCredentialsParameters oauthClientCredentialsModel =
             (OAuthClientCredentialsParameters) authenticationParameters;

polaris-core/src/main/java/org/apache/polaris/core/connection/AuthenticationType.java

@@ -35,6 +35,7 @@ public enum AuthenticationType {
   BEARER(2),
   IMPLICIT(3),
   SIGV4(4),
+  GCP(5),
   ;
 
   private static final AuthenticationType[] REVERSE_MAPPING_ARRAY;

polaris-core/src/main/java/org/apache/polaris/core/connection/ConnectionConfigInfoDpo.java

@@ -205,7 +205,8 @@ public static ConnectionConfigInfoDpo fromConnectionConfigInfoModelWithSecrets(
                 icebergRestConfigModel.getUri(),
                 authenticationParameters,
                 null /*Service Identity Info*/,
-                icebergRestConfigModel.getRemoteCatalogName());
+                icebergRestConfigModel.getRemoteCatalogName(),
+                icebergRestConfigModel.getProperties());
         break;
       case HADOOP:
         HadoopConnectionConfigInfo hadoopConfigModel =

polaris-core/src/main/java/org/apache/polaris/core/connection/GcpAuthenticationParametersDpo.java

@@ -0,0 +1,72 @@
+/*
+ * 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.polaris.core.connection;
+
+import jakarta.annotation.Nonnull;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.iceberg.rest.auth.AuthProperties;
+import org.apache.polaris.core.admin.model.AuthenticationParameters;
+import org.apache.polaris.core.admin.model.GcpAuthenticationParameters;
+import org.apache.polaris.core.credentials.PolarisCredentialManager;
+
+/**
+ * See {@link org.apache.iceberg.rest.RESTUtil#configHeaders(Map)} and {@link
+ * org.apache.iceberg.rest.auth.AuthManagers#loadAuthManager(String, Map)} for why we do this.
+ */
+public class GcpAuthenticationParametersDpo extends AuthenticationParametersDpo {
+
+  public GcpAuthenticationParametersDpo() {
+    super(AuthenticationType.GCP.getCode());
+  }
+
+  @Nonnull
+  @Override
+  public Map<String, String> asIcebergCatalogProperties(
+      PolarisCredentialManager credentialManager) {
+    HashMap<String, String> properties = new HashMap<>();
+    properties.put(AuthProperties.AUTH_TYPE, AuthProperties.AUTH_TYPE_GOOGLE);
+    return properties;
+  }
+
+  @Nonnull
+  @Override
+  public GcpAuthenticationParameters asAuthenticationParametersModel() {
+    return GcpAuthenticationParameters.builder()
+        .setAuthenticationType(AuthenticationParameters.AuthenticationTypeEnum.GCP)
+        .build();
+  }
+
+  @Override
+  public String toString() {
+    return "GcpAuthenticationParametersDpo{}";
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (o == null || !(o instanceof GcpAuthenticationParametersDpo that)) return false;
+    return true;
+  }
+
+  @Override
+  public int hashCode() {
+    return -1;
+  }
+}

polaris-core/src/main/java/org/apache/polaris/core/connection/iceberg/IcebergRestConnectionConfigInfoDpo.java

@@ -23,6 +23,7 @@
 import jakarta.annotation.Nonnull;
 import jakarta.annotation.Nullable;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import org.apache.iceberg.CatalogProperties;
@@ -43,18 +44,32 @@
 public class IcebergRestConnectionConfigInfoDpo extends ConnectionConfigInfoDpo
     implements IcebergCatalogPropertiesProvider {
 
+  public static final String GOOGLE_USER_PROJECT_HEADER_KEY = "header.x-goog-user-project";
+
+  private static final List<String> ALLOWED_PROPERTIES = List.of(GOOGLE_USER_PROJECT_HEADER_KEY);
+
   private final String remoteCatalogName;
 
+  /**
+   * @param properties Properties that might be specifically needed for a particular implementation
+   *     of a REST API.
+   */
   public IcebergRestConnectionConfigInfoDpo(
       @JsonProperty(value = "uri", required = true) @Nonnull String uri,
       @JsonProperty(value = "authenticationParameters", required = true) @Nonnull
           AuthenticationParametersDpo authenticationParameters,
       @JsonProperty(value = "serviceIdentity", required = false) @Nullable
           ServiceIdentityInfoDpo serviceIdentityInfo,
       @JsonProperty(value = "remoteCatalogName", required = false) @Nullable
-          String remoteCatalogName) {
+          String remoteCatalogName,
+      @JsonProperty(value = "properties", required = false) @Nullable
+          Map<String, String> properties) {
     super(
-        ConnectionType.ICEBERG_REST.getCode(), uri, authenticationParameters, serviceIdentityInfo);
+        ConnectionType.ICEBERG_REST.getCode(),
+        uri,
+        authenticationParameters,
+        serviceIdentityInfo,
+        properties);
     this.remoteCatalogName = remoteCatalogName;
   }
 
@@ -72,6 +87,13 @@ public String getRemoteCatalogName() {
     }
     // Add authentication-specific metadata (non-credential properties)
     properties.putAll(getAuthenticationParameters().asIcebergCatalogProperties(credentialManager));
+
+    for (String headerKey : ALLOWED_PROPERTIES) {
+      if (getProperties().containsKey(headerKey)) {
+        properties.put(headerKey, getProperties().get(headerKey));
+      }
+    }
+
     // Add connection credentials from Polaris credential manager
     ConnectionCredentials connectionCredentials = credentialManager.getConnectionCredentials(this);
     properties.putAll(connectionCredentials.credentials());
@@ -82,7 +104,11 @@ public String getRemoteCatalogName() {
   public ConnectionConfigInfoDpo withServiceIdentity(
       @Nonnull ServiceIdentityInfoDpo serviceIdentityInfo) {
     return new IcebergRestConnectionConfigInfoDpo(
-        getUri(), getAuthenticationParameters(), serviceIdentityInfo, getRemoteCatalogName());
+        getUri(),
+        getAuthenticationParameters(),
+        serviceIdentityInfo,
+        getRemoteCatalogName(),
+        getProperties());
   }
 
   @Override
@@ -100,6 +126,7 @@ public ConnectionConfigInfo asConnectionConfigInfoModel(
                     serviceIdentityInfoDpo ->
                         serviceIdentityInfoDpo.asServiceIdentityInfoModel(serviceIdentityProvider))
                 .orElse(null))
+        .setProperties(getProperties())
         .build();
   }
 
@@ -111,6 +138,7 @@ public String toString() {
         .add("remoteCatalogName", getRemoteCatalogName())
         .add("authenticationParameters", getAuthenticationParameters().toString())
         .add("serviceIdentity", getServiceIdentity())
+        .add("properties", getProperties())
         .toString();
   }
 }

polaris-core/src/test/java/org/apache/polaris/core/connection/GcpAuthenticationParametersDpoTest.java

@@ -0,0 +1,82 @@
+/*
+ * 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.polaris.core.connection;
+
+import static java.util.stream.Collectors.toUnmodifiableSet;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCharSequence;
+
+import java.lang.reflect.Method;
+import java.util.Arrays;
+import java.util.Map;
+import java.util.Set;
+import org.apache.polaris.core.admin.model.GcpAuthenticationParameters;
+import org.apache.polaris.core.connection.iceberg.IcebergRestConnectionConfigInfoDpo;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+class GcpAuthenticationParametersDpoTest {
+
+  private GcpAuthenticationParametersDpo dpo;
+
+  @BeforeEach
+  void setUp() {
+    dpo = new GcpAuthenticationParametersDpo();
+  }
+
+  @Test
+  void testSerializeAndDeserialize() throws Exception {
+    var connectionConfig =
+        new IcebergRestConnectionConfigInfoDpo(
+            "https://biglake.googleapis.com/iceberg/v1/restcatalog",
+            dpo,
+            null,
+            null,
+            Map.of("x", "y"));
+
+    assertThatCharSequence(connectionConfig.toString())
+        .isEqualTo(ConnectionConfigInfoDpo.deserialize(connectionConfig.serialize()).toString());
+  }
+
+  @Test
+  void testConversionToDTOCapturesAllFields() {
+    GcpAuthenticationParameters authenticationParameters = dpo.asAuthenticationParametersModel();
+    Set<String> dtoGetMethods =
+        Arrays.stream(GcpAuthenticationParameters.class.getDeclaredMethods())
+            .map(Method::getName)
+            .filter(x -> x.startsWith("get"))
+            .collect(toUnmodifiableSet());
+    dtoGetMethods.stream()
+        .forEach(
+            x -> {
+              try {
+                var expected =
+                    GcpAuthenticationParametersDpo.class.getMethod(x, (Class<?>) null).invoke(dpo);
+                var actual =
+                    GcpAuthenticationParameters.class
+                        .getMethod(x, (Class<?>) null)
+                        .invoke(authenticationParameters);
+                assertThat(expected).isEqualTo(actual);
+              } catch (Exception e) {
+                throw new RuntimeException(e);
+              }
+            });
+  }
+}

polaris-core/src/test/java/org/apache/polaris/core/connection/iceberg/IcebergRestConnectionConfigInfoDpoTest.java

@@ -0,0 +1,70 @@
+/*
+ * 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.polaris.core.connection.iceberg;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCharSequence;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.util.Map;
+import org.apache.polaris.core.admin.model.ConnectionConfigInfo;
+import org.apache.polaris.core.connection.ConnectionConfigInfoDpo;
+import org.apache.polaris.core.connection.GcpAuthenticationParametersDpo;
+import org.apache.polaris.core.credentials.PolarisCredentialManager;
+import org.apache.polaris.core.credentials.connection.ConnectionCredentials;
+import org.jspecify.annotations.NonNull;
+import org.junit.jupiter.api.Test;
+
+class IcebergRestConnectionConfigInfoDpoTest {
+
+  @Test
+  void testRoundTrip() {
+    IcebergRestConnectionConfigInfoDpo dpo = createDpo(Map.of("x", "y"));
+    ConnectionConfigInfo dto = dpo.asConnectionConfigInfoModel(null);
+    assertThatCharSequence(dpo.toString())
+        .isEqualTo(
+            ConnectionConfigInfoDpo.fromConnectionConfigInfoModelWithSecrets(dto, Map.of())
+                .toString());
+  }
+
+  @Test
+  void testNullAdditionalHeadersHandledGracefully() {
+    IcebergRestConnectionConfigInfoDpo dpo = createDpo(null);
+    PolarisCredentialManager mockCredentialManager = mock(PolarisCredentialManager.class);
+    ConnectionCredentials mockCredentials = mock(ConnectionCredentials.class);
+    String expectedKey = "credential_key";
+    String expectedValue = "credential_value";
+    when(mockCredentials.credentials()).thenReturn(Map.of(expectedKey, expectedValue));
+    when(mockCredentialManager.getConnectionCredentials(dpo)).thenReturn(mockCredentials);
+    Map<String, String> properties = dpo.asIcebergCatalogProperties(mockCredentialManager);
+    assertThat(properties).containsEntry(expectedKey, expectedValue);
+  }
+
+  private static @NonNull IcebergRestConnectionConfigInfoDpo createDpo(
+      Map<String, String> additionalHeaders) {
+    return new IcebergRestConnectionConfigInfoDpo(
+        "https://biglake.googleapis.com/iceberg/v1/restcatalog",
+        new GcpAuthenticationParametersDpo(),
+        null,
+        null,
+        additionalHeaders);
+  }
+}

runtime/server/build.gradle.kts

@@ -48,7 +48,10 @@ dependencies {
   }
 
   // enforce the Quarkus _platform_ here, to get a consistent and validated set of dependencies
-  implementation(enforcedPlatform(libs.quarkus.bom))
+  implementation(enforcedPlatform(libs.quarkus.bom)) {
+    exclude(group = "com.google.protobuf", module = "protobuf-java")
+    exclude(group = "com.google.protobuf", module = "protobuf-java-util")
+  }
   implementation("io.quarkus:quarkus-container-image-docker")
 }