support credential cache

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

@@ -28,7 +28,7 @@
 public class ADLSTokenCredential implements Credential {
 
   /** ADLS SAS token credential type. */
-  public static final String ADLS_SAS_TOKEN_CREDENTIAL_TYPE = "adls-sas-token";
+  public static final String ADLS_SAS_TOKEN_CREDENTIAL_TYPE = "adls-token";
   /** ADLS base domain */
   public static final String ADLS_DOMAIN = "dfs.core.windows.net";
   /** ADLS storage account name */

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

@@ -22,6 +22,9 @@
 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_MAX_TIME_IN_SECS =
+      "credential-cache-expire-time-in-secs";
+  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/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

@@ -19,10 +19,10 @@
 
 package org.apache.gravitino.credential;
 
-import com.google.common.base.Preconditions;
 import java.io.Closeable;
 import java.io.IOException;
 import java.util.Map;
+import org.apache.gravitino.exceptions.NoSuchCredentialException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -34,20 +34,21 @@ public class CatalogCredentialManager implements Closeable {
 
   private static final Logger LOG = LoggerFactory.getLogger(CatalogCredentialManager.class);
 
+  private CredentialCache<CredentialCacheKey> cacheManager;
+
   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.cacheManager = new CredentialCache();
+    cacheManager.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 cacheManager.getCredential(credentialCacheKey, cacheKey -> getCredential(cacheKey));
   }
 
   @Override
@@ -67,4 +68,15 @@ public void close() {
               }
             });
   }
+
+  private Credential getCredential(CredentialCacheKey credentialCacheKey) {
+    String credentialType = credentialCacheKey.getCredentialType();
+    CredentialContext context = credentialCacheKey.getCredentialContext();
+    LOG.info("try get credential, credential type: {}, context: {}", credentialType, context);
+    CredentialProvider credentialProvider = credentialProviders.get(credentialType);
+    if (credentialProvider == null) {
+      throw new NoSuchCredentialException("No such credential: %s", credentialType);
+    }
+    return credentialProvider.getCredential(context);
+  }
 }

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

@@ -0,0 +1,104 @@
+/*
+ *  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.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
+import org.apache.gravitino.credential.config.CredentialConfig;
+import org.checkerframework.checker.index.qual.NonNegative;
+import org.checkerframework.checker.nullness.qual.NonNull;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class CredentialCache<T> {
+
+  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 long credentialMaxCacheTimeInMs;
+
+    public CredentialExpireTimeCalculator(long credentialMaxCacheTimeInSecs) {
+      this.credentialMaxCacheTimeInMs = credentialMaxCacheTimeInSecs * 1000;
+    }
+
+    // Set expire time after add a credential in the cache.
+    @Override
+    public long expireAfterCreate(
+        @NonNull T key, @NonNull Credential credential, long currentTime) {
+      long credentialExpireTime = credential.expireTimeInMs();
+      long timeToExpire = credentialExpireTime - System.currentTimeMillis();
+      if (timeToExpire <= 0) {
+        return 0;
+      }
+
+      long idealExpireTime = Math.min(timeToExpire / 2, credentialMaxCacheTimeInMs);
+      LOG.info("credential expire time: {}", idealExpireTime);
+      return TimeUnit.MILLISECONDS.toNanos(idealExpireTime);
+    }
+
+    // not change expire time after update credential, this should not happen.
+    @Override
+    public long expireAfterUpdate(
+        @NonNull T key,
+        @NonNull Credential value,
+        long currentTime,
+        @NonNegative long currentDuration) {
+      return currentDuration;
+    }
+
+    // not change expire time after read credential.
+    @Override
+    public long expireAfterRead(
+        @NonNull T key,
+        @NonNull Credential value,
+        long currentTime,
+        @NonNegative long currentDuration) {
+      return currentDuration;
+    }
+  }
+
+  private Cache<T, Credential> credentialCache;
+
+  public void initialize(Map<String, String> catalogProperties) {
+    CredentialConfig credentialConfig = new CredentialConfig(catalogProperties);
+    long cache_size = credentialConfig.get(CredentialConfig.CREDENTIAL_MAX_CACHE_SIZE);
+    long cache_time = credentialConfig.get(CredentialConfig.CREDENTIAL_MAX_CACHE_TIME);
+
+    this.credentialCache =
+        Caffeine.newBuilder()
+            .expireAfter(new CredentialExpireTimeCalculator(cache_time))
+            .maximumSize(cache_size)
+            .removalListener(
+                (cacheKey, credential, c) -> {
+                  LOG.info("credential expire {}.", cacheKey);
+                })
+            .build();
+  }
+
+  public Credential getCredential(T cacheKey, Function<T, Credential> credentialSupplier) {
+    return credentialCache.get(cacheKey, key -> credentialSupplier.apply(cacheKey));
+  }
+}

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 String credentialType;
+  private 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/CredentialContext.java

@@ -21,6 +21,7 @@
 
 /** Contains credential context information to get credential from a credential provider. */
 public interface CredentialContext {
+
   /**
    * Providing the username.
    *

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();
+  }
 }