[#4992] support credential vending framework

api/src/main/java/org/apache/gravitino/credential/Credential.java

@@ -0,0 +1,67 @@
+/*
+ *  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.gravitino.credential;
+
+import com.google.common.collect.ImmutableMap;
+import java.util.Map;
+
+/** Interface representing a credential with type, expiration time, and additional information. */
+public interface Credential {
+  /** Credential type in the credential. */
+  String CREDENTIAL_TYPE = "credential-type";
+  /** Credential expire time in ms since the epoch. */
+  String EXPIRE_TIME_IN_MS = "expire-time-in-ms";
+
+  /**
+   * Returns the type of the credential. It should be the same as the credential type of the
+   * credential provider.
+   *
+   * @return the credential type as a String.
+   */
+  String credentialType();
+
+  /**
+   * Returns the expiration time of the credential in milliseconds since the epoch, 0 means not
+   * expire.
+   *
+   * @return the expiration time as a long.
+   */
+  long expireTimeInMs();
+
+  /**
+   * Returns credential information.
+   *
+   * @return a map of credential information.
+   */
+  Map<String, String> credentialInfo();
+
+  /**
+   * Converts the credential to properties to transfer the credential though API.
+   *
+   * @return a map containing credential properties.
+   */
+  default Map<String, String> toProperties() {
+    return new ImmutableMap.Builder<String, String>()
+        .putAll(credentialInfo())
+        .put(CREDENTIAL_TYPE, credentialType())
+        .put(EXPIRE_TIME_IN_MS, String.valueOf(expireTimeInMs()))
+        .build();
+  }
+}

core/src/main/java/org/apache/gravitino/credential/CatalogCredentialContext.java

@@ -0,0 +1,38 @@
+/*
+ *  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.gravitino.credential;
+
+import com.google.common.base.Preconditions;
+import javax.validation.constraints.NotNull;
+
+/** CatalogCredentialContext is generated when user requesting catalog credentials. */
+public class CatalogCredentialContext implements CredentialContext {
+  @NotNull private final String userName;
+
+  public CatalogCredentialContext(String userName) {
+    Preconditions.checkNotNull(userName, "User name should not be null");
+    this.userName = userName;
+  }
+
+  @Override
+  public String getUserName() {
+    return userName;
+  }
+}

core/src/main/java/org/apache/gravitino/credential/CredentialContext.java

@@ -0,0 +1,30 @@
+/*
+ *  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.gravitino.credential;
+
+/** Contains credential context information to get credential from a credential provider. */
+public interface CredentialContext {
+  /**
+   * Providing the username.
+   *
+   * @return A string identifying user name.
+   */
+  String getUserName();
+}

core/src/main/java/org/apache/gravitino/credential/CredentialProvider.java

@@ -0,0 +1,56 @@
+/*
+ *  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.gravitino.credential;
+
+import java.io.Closeable;
+import java.util.Map;
+import javax.annotation.Nullable;
+
+/**
+ * Interface for credential providers.
+ *
+ * <p>A credential provider is responsible for managing and retrieving credentials.
+ */
+public interface CredentialProvider extends Closeable {
+  /**
+   * Initializes the credential provider with catalog properties.
+   *
+   * @param properties catalog properties that can be used to configure the provider. The specific
+   *     properties required vary by implementation.
+   */
+  void initialize(Map<String, String> properties);
+
+  /**
+   * Returns the type of credential, it should be identical in Gravitino.
+   *
+   * @return A string identifying the type of credentials.
+   */
+  String credentialType();
+
+  /**
+   * Obtains a credential based on the provided context information.
+   *
+   * @param context A context object providing necessary information for retrieving credentials.
+   * @return A Credential object containing the authentication information needed to access a system
+   *     or resource. Null will be returned if no credential is available.
+   */
+  @Nullable
+  Credential getCredential(CredentialContext context);
+}

core/src/main/java/org/apache/gravitino/credential/CredentialProviderFactory.java

@@ -0,0 +1,69 @@
+/*
+ *  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.gravitino.credential;
+
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Streams;
+import java.util.List;
+import java.util.Map;
+import java.util.ServiceLoader;
+import java.util.stream.Collectors;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class CredentialProviderFactory {
+  private static final Logger LOG = LoggerFactory.getLogger(CredentialProviderFactory.class);
+
+  public static CredentialProvider create(
+      String credentialType, Map<String, String> catalogProperties) {
+    Class<? extends CredentialProvider> providerClz = lookupCredentialProvider(credentialType);
+    try {
+      CredentialProvider provider = providerClz.getDeclaredConstructor().newInstance();
+      provider.initialize(catalogProperties);
+      return provider;
+    } catch (Exception e) {
+      LOG.warn("Create credential provider failed, {}", credentialType, e);
+      throw new RuntimeException(e);
+    }
+  }
+
+  private static Class<? extends CredentialProvider> lookupCredentialProvider(
+      String credentialType) {
+    ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
+    ServiceLoader<CredentialProvider> serviceLoader =
+        ServiceLoader.load(CredentialProvider.class, classLoader);
+    List<Class<? extends CredentialProvider>> providers =
+        Streams.stream(serviceLoader.iterator())
+            .filter(
+                credentialProvider ->
+                    credentialType.equalsIgnoreCase(credentialProvider.credentialType()))
+            .map(CredentialProvider::getClass)
+            .collect(Collectors.toList());
+
+    if (providers.isEmpty()) {
+      throw new IllegalArgumentException("No credential provider found for: " + credentialType);
+    } else if (providers.size() > 1) {
+      throw new IllegalArgumentException(
+          "Multiple credential providers found for: " + credentialType);
+    } else {
+      return Iterables.getOnlyElement(providers);
+    }
+  }
+}

core/src/main/java/org/apache/gravitino/credential/PathBasedCredentialContext.java

@@ -0,0 +1,58 @@
+/*
+ *  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.gravitino.credential;
+
+import com.google.common.base.Preconditions;
+import java.util.Set;
+import javax.validation.constraints.NotNull;
+
+/**
+ * LocationContext is generated when user requesting resources associated with storage location like
+ * table, fileset, etc.
+ */
+public class PathBasedCredentialContext implements CredentialContext {
+
+  @NotNull private final Set<String> writePaths;
+  @NotNull private final Set<String> readPaths;
+  @NotNull private final String userName;
+
+  public PathBasedCredentialContext(
+      String userName, Set<String> writePaths, Set<String> readPaths) {
+    Preconditions.checkNotNull(userName, "User name should not be null");
+    Preconditions.checkNotNull(writePaths, "Write paths should not be null");
+    Preconditions.checkNotNull(readPaths, "Read paths should not be null");
+    this.userName = userName;
+    this.writePaths = writePaths;
+    this.readPaths = readPaths;
+  }
+
+  @Override
+  public String getUserName() {
+    return userName;
+  }
+
+  public Set<String> getWritePaths() {
+    return writePaths;
+  }
+
+  public Set<String> getReadPaths() {
+    return readPaths;
+  }
+}