[apache#4398] feat(core): support credential cache for Gravitino serv…

…er (apache#5995)

### What changes were proposed in this pull request?

add credential cache for Gravitino server, not support for Iceberg rest
server yet.

### Why are the changes needed?

Fix: apache#4398 

### Does this PR introduce _any_ user-facing change?
no

### How was this patch tested?

testing in local env, get credential from Gravitino server and see
whether it's fetched from remote or local cache

catalogs/catalog-common/src/main/java/org/apache/gravitino/credential/CredentialConstants.java

@@ -22,6 +22,8 @@
 public class CredentialConstants {
   public static final String CREDENTIAL_PROVIDER_TYPE = "credential-provider-type";
   public static final String CREDENTIAL_PROVIDERS = "credential-providers";
+  public static final String CREDENTIAL_CACHE_EXPIRE_RATIO = "credential-cache-expire-ratio";
+  public static final String CREDENTIAL_CACHE_MAX_SIZE = "credential-cache-max-size";
   public static final String S3_TOKEN_CREDENTIAL_PROVIDER = "s3-token";
   public static final String S3_TOKEN_EXPIRE_IN_SECS = "s3-token-expire-in-secs";
 

core/src/main/java/org/apache/gravitino/config/ConfigBuilder.java

@@ -170,6 +170,31 @@ public ConfigEntry<Long> longConf() {
     return conf;
   }
 
+  /**
+   * Creates a configuration entry for Double data type.
+   *
+   * @return The created ConfigEntry instance for Double data type.
+   */
+  public ConfigEntry<Double> doubleConf() {
+    ConfigEntry<Double> conf =
+        new ConfigEntry<>(key, version, doc, alternatives, isPublic, isDeprecated);
+    Function<String, Double> func =
+        s -> {
+          if (s == null || s.isEmpty()) {
+            return null;
+          } else {
+            return Double.parseDouble(s);
+          }
+        };
+    conf.setValueConverter(func);
+
+    Function<Double, String> stringFunc =
+        t -> Optional.ofNullable(t).map(String::valueOf).orElse(null);
+    conf.setStringConverter(stringFunc);
+
+    return conf;
+  }
+
   /**
    * Creates a configuration entry for Boolean data type.
    *

core/src/main/java/org/apache/gravitino/connector/PropertyEntry.java

@@ -29,6 +29,7 @@
 
 @Getter
 public final class PropertyEntry<T> {
+
   private final String name;
   private final String description;
   private final boolean required;
@@ -90,6 +91,7 @@ private PropertyEntry(
   }
 
   public static class Builder<T> {
+
     private String name;
     private String description;
     private boolean required;
@@ -214,6 +216,28 @@ public static PropertyEntry<Long> longPropertyEntry(
         .build();
   }
 
+  public static PropertyEntry<Double> doublePropertyEntry(
+      String name,
+      String description,
+      boolean required,
+      boolean immutable,
+      double defaultValue,
+      boolean hidden,
+      boolean reserved) {
+    return new Builder<Double>()
+        .withName(name)
+        .withDescription(description)
+        .withRequired(required)
+        .withImmutable(immutable)
+        .withJavaType(Double.class)
+        .withDefaultValue(defaultValue)
+        .withDecoder(Double::parseDouble)
+        .withEncoder(String::valueOf)
+        .withHidden(hidden)
+        .withReserved(reserved)
+        .build();
+  }
+
   public static PropertyEntry<Integer> integerPropertyEntry(
       String name,
       String description,

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

@@ -19,6 +19,7 @@
 
 package org.apache.gravitino.credential;
 
+import com.google.common.base.Objects;
 import com.google.common.base.Preconditions;
 import javax.validation.constraints.NotNull;
 
@@ -35,4 +36,27 @@ public CatalogCredentialContext(String userName) {
   public String getUserName() {
     return userName;
   }
+
+  @Override
+  public int hashCode() {
+    return Objects.hashCode(userName);
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || !(o instanceof CatalogCredentialContext)) {
+      return false;
+    }
+    return Objects.equal(userName, ((CatalogCredentialContext) o).userName);
+  }
+
+  @Override
+  public String toString() {
+    StringBuilder stringBuilder = new StringBuilder();
+    stringBuilder.append("User name: ").append(userName);
+    return stringBuilder.toString();
+  }
 }

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

@@ -34,20 +34,21 @@ public class CatalogCredentialManager implements Closeable {
 
   private static final Logger LOG = LoggerFactory.getLogger(CatalogCredentialManager.class);
 
+  private final CredentialCache<CredentialCacheKey> credentialCache;
+
   private final String catalogName;
   private final Map<String, CredentialProvider> credentialProviders;
 
   public CatalogCredentialManager(String catalogName, Map<String, String> catalogProperties) {
     this.catalogName = catalogName;
     this.credentialProviders = CredentialUtils.loadCredentialProviders(catalogProperties);
+    this.credentialCache = new CredentialCache();
+    credentialCache.initialize(catalogProperties);
   }
 
   public Credential getCredential(String credentialType, CredentialContext context) {
-    // todo: add credential cache
-    Preconditions.checkState(
-        credentialProviders.containsKey(credentialType),
-        String.format("Credential %s not found", credentialType));
-    return credentialProviders.get(credentialType).getCredential(context);
+    CredentialCacheKey credentialCacheKey = new CredentialCacheKey(credentialType, context);
+    return credentialCache.getCredential(credentialCacheKey, cacheKey -> doGetCredential(cacheKey));
   }
 
   @Override
@@ -67,4 +68,14 @@ public void close() {
               }
             });
   }
+
+  private Credential doGetCredential(CredentialCacheKey credentialCacheKey) {
+    String credentialType = credentialCacheKey.getCredentialType();
+    CredentialContext context = credentialCacheKey.getCredentialContext();
+    LOG.debug("Try get credential, credential type: {}, context: {}.", credentialType, context);
+    Preconditions.checkState(
+        credentialProviders.containsKey(credentialType),
+        String.format("Credential %s not found", credentialType));
+    return credentialProviders.get(credentialType).getCredential(context);
+  }
 }

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

@@ -0,0 +1,101 @@
+/*
+ *  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.github.benmanes.caffeine.cache.Cache;
+import com.github.benmanes.caffeine.cache.Caffeine;
+import com.github.benmanes.caffeine.cache.Expiry;
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
+import org.apache.gravitino.credential.config.CredentialConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class CredentialCache<T> implements Closeable {
+
+  private static final Logger LOG = LoggerFactory.getLogger(CredentialCache.class);
+
+  // Calculates the credential expire time in the cache.
+  static class CredentialExpireTimeCalculator<T> implements Expiry<T, Credential> {
+
+    private double credentialCacheExpireRatio;
+
+    public CredentialExpireTimeCalculator(double credentialCacheExpireRatio) {
+      this.credentialCacheExpireRatio = credentialCacheExpireRatio;
+    }
+
+    // Set expire time after add a credential in the cache.
+    @Override
+    public long expireAfterCreate(T key, Credential credential, long currentTime) {
+      long credentialExpireTime = credential.expireTimeInMs();
+      long timeToExpire = credentialExpireTime - System.currentTimeMillis();
+      if (timeToExpire <= 0) {
+        return 0;
+      }
+
+      timeToExpire = (long) (timeToExpire * credentialCacheExpireRatio);
+      return TimeUnit.MILLISECONDS.toNanos(timeToExpire);
+    }
+
+    // Not change expire time after update credential, this should not happen.
+    @Override
+    public long expireAfterUpdate(T key, Credential value, long currentTime, long currentDuration) {
+      return currentDuration;
+    }
+
+    // Not change expire time after read credential.
+    @Override
+    public long expireAfterRead(T key, Credential value, long currentTime, long currentDuration) {
+      return currentDuration;
+    }
+  }
+
+  private Cache<T, Credential> credentialCache;
+
+  public void initialize(Map<String, String> catalogProperties) {
+    CredentialConfig credentialConfig = new CredentialConfig(catalogProperties);
+    long cacheSize = credentialConfig.get(CredentialConfig.CREDENTIAL_CACHE_MAX_SIZE);
+    double cacheExpireRatio = credentialConfig.get(CredentialConfig.CREDENTIAL_CACHE_EXPIRE_RATIO);
+
+    this.credentialCache =
+        Caffeine.newBuilder()
+            .expireAfter(new CredentialExpireTimeCalculator(cacheExpireRatio))
+            .maximumSize(cacheSize)
+            .removalListener(
+                (cacheKey, credential, c) ->
+                    LOG.debug("Credential expire, cache key: {}.", cacheKey))
+            .build();
+  }
+
+  public Credential getCredential(T cacheKey, Function<T, Credential> credentialSupplier) {
+    return credentialCache.get(cacheKey, key -> credentialSupplier.apply(cacheKey));
+  }
+
+  @Override
+  public void close() throws IOException {
+    if (credentialCache != null) {
+      credentialCache.invalidateAll();
+      credentialCache = null;
+    }
+  }
+}

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

@@ -0,0 +1,64 @@
+/*
+ *  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.util.Objects;
+import lombok.Getter;
+
+@Getter
+public class CredentialCacheKey {
+
+  private final String credentialType;
+  private final CredentialContext credentialContext;
+
+  public CredentialCacheKey(String credentialType, CredentialContext credentialContext) {
+    this.credentialType = credentialType;
+    this.credentialContext = credentialContext;
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(credentialType, credentialContext);
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || !(o instanceof CredentialCacheKey)) {
+      return false;
+    }
+    CredentialCacheKey that = (CredentialCacheKey) o;
+    return Objects.equals(credentialType, that.credentialType)
+        && Objects.equals(credentialContext, that.credentialContext);
+  }
+
+  @Override
+  public String toString() {
+    StringBuilder stringBuilder = new StringBuilder();
+    stringBuilder
+        .append("credentialType: ")
+        .append(credentialType)
+        .append("credentialContext: ")
+        .append(credentialContext);
+    return stringBuilder.toString();
+  }
+}

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

@@ -20,6 +20,7 @@
 package org.apache.gravitino.credential;
 
 import com.google.common.base.Preconditions;
+import java.util.Objects;
 import java.util.Set;
 import javax.validation.constraints.NotNull;
 
@@ -55,4 +56,36 @@ public Set<String> getWritePaths() {
   public Set<String> getReadPaths() {
     return readPaths;
   }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(userName, writePaths, readPaths);
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || !(o instanceof PathBasedCredentialContext)) {
+      return false;
+    }
+    PathBasedCredentialContext that = (PathBasedCredentialContext) o;
+    return Objects.equals(userName, that.userName)
+        && Objects.equals(writePaths, that.writePaths)
+        && Objects.equals(readPaths, that.readPaths);
+  }
+
+  @Override
+  public String toString() {
+    StringBuilder stringBuilder = new StringBuilder();
+    stringBuilder
+        .append("User name: ")
+        .append(userName)
+        .append(", write path: ")
+        .append(writePaths)
+        .append(", read path: ")
+        .append(readPaths);
+    return stringBuilder.toString();
+  }
 }