[#5842] feat(core): supports credential REST endpoint in Gravitino server

core/src/main/java/org/apache/gravitino/GravitinoEnv.java

@@ -28,6 +28,7 @@
 import org.apache.gravitino.catalog.CatalogDispatcher;
 import org.apache.gravitino.catalog.CatalogManager;
 import org.apache.gravitino.catalog.CatalogNormalizeDispatcher;
+import org.apache.gravitino.catalog.CredentialManager;
 import org.apache.gravitino.catalog.FilesetDispatcher;
 import org.apache.gravitino.catalog.FilesetNormalizeDispatcher;
 import org.apache.gravitino.catalog.FilesetOperationDispatcher;
@@ -105,6 +106,8 @@ public class GravitinoEnv {
 
   private MetalakeDispatcher metalakeDispatcher;
 
+  private CredentialManager credentialManager;
+
   private AccessControlDispatcher accessControlDispatcher;
 
   private IdGenerator idGenerator;
@@ -257,6 +260,15 @@ public MetalakeDispatcher metalakeDispatcher() {
     return metalakeDispatcher;
   }
 
+  /**
+   * Get the {@link CredentialManager} associated with the Gravitino environment.
+   *
+   * @return The {@link CredentialManager} instance.
+   */
+  public CredentialManager credentialManager() {
+    return credentialManager;
+  }
+
   /**
    * Get the IdGenerator associated with the Gravitino environment.
    *
@@ -417,6 +429,8 @@ private void initGravitinoServerComponents() {
         new CatalogNormalizeDispatcher(catalogHookDispatcher);
     this.catalogDispatcher = new CatalogEventDispatcher(eventBus, catalogNormalizeDispatcher);
 
+    this.credentialManager = new CredentialManager(catalogManager, entityStore, idGenerator);
+
     SchemaOperationDispatcher schemaOperationDispatcher =
         new SchemaOperationDispatcher(catalogManager, entityStore, idGenerator);
     SchemaHookDispatcher schemaHookDispatcher = new SchemaHookDispatcher(schemaOperationDispatcher);

core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java

@@ -133,6 +133,10 @@ public CatalogWrapper(BaseCatalog catalog, IsolatedClassLoader classLoader) {
       this.classLoader = classLoader;
     }
 
+    public BaseCatalog catalog() {
+      return catalog;
+    }
+
     public <R> R doWithSchemaOps(ThrowableFunction<SupportsSchemas, R> fn) throws Exception {
       return classLoader.withClassLoader(
           cl -> {

core/src/main/java/org/apache/gravitino/catalog/CredentialManager.java

@@ -0,0 +1,53 @@
+/*
+ *  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.catalog;
+
+import java.util.List;
+import org.apache.commons.lang3.NotImplementedException;
+import org.apache.gravitino.EntityStore;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.connector.BaseCatalog;
+import org.apache.gravitino.credential.Credential;
+import org.apache.gravitino.exceptions.NoSuchCatalogException;
+import org.apache.gravitino.storage.IdGenerator;
+import org.apache.gravitino.utils.NameIdentifierUtil;
+
+/** Get credentials with the specific catalog classloader. */
+public class CredentialManager extends OperationDispatcher {
+
+  public CredentialManager(
+      CatalogManager catalogManager, EntityStore store, IdGenerator idGenerator) {
+    super(catalogManager, store, idGenerator);
+  }
+
+  public List<Credential> getCredentials(NameIdentifier identifier) {
+    return doWithCatalog(
+        NameIdentifierUtil.getCatalogIdentifier(identifier),
+        c -> getCredentials(c.catalog(), identifier),
+        NoSuchCatalogException.class);
+  }
+
+  private List<Credential> getCredentials(BaseCatalog catalog, NameIdentifier identifier) {
+    throw new NotImplementedException(
+        String.format(
+            "Load credentials is not implemented for catalog: %s, identifier: %s",
+            catalog.name(), identifier));
+  }
+}

server/src/main/java/org/apache/gravitino/server/GravitinoServer.java

@@ -26,6 +26,7 @@
 import org.apache.gravitino.Configs;
 import org.apache.gravitino.GravitinoEnv;
 import org.apache.gravitino.catalog.CatalogDispatcher;
+import org.apache.gravitino.catalog.CredentialManager;
 import org.apache.gravitino.catalog.FilesetDispatcher;
 import org.apache.gravitino.catalog.PartitionDispatcher;
 import org.apache.gravitino.catalog.SchemaDispatcher;
@@ -114,6 +115,7 @@ protected void configure() {
             bind(gravitinoEnv.filesetDispatcher()).to(FilesetDispatcher.class).ranked(1);
             bind(gravitinoEnv.topicDispatcher()).to(TopicDispatcher.class).ranked(1);
             bind(gravitinoEnv.tagManager()).to(TagManager.class).ranked(1);
+            bind(gravitinoEnv.credentialManager()).to(CredentialManager.class).ranked(1);
           }
         });
     register(JsonProcessingExceptionMapper.class);

server/src/main/java/org/apache/gravitino/server/web/rest/ExceptionHandlers.java

@@ -121,6 +121,11 @@ public static Response handleTagException(
     return TagExceptionHandler.INSTANCE.handle(op, tag, parent, e);
   }
 
+  public static Response handleCredentialException(
+      OperationType op, String metadataObjectName, Exception e) {
+    return CredentialExceptionHandler.INSTANCE.handle(op, metadataObjectName, "", e);
+  }
+
   public static Response handleTestConnectionException(Exception e) {
     ErrorResponse response;
     if (e instanceof IllegalArgumentException) {
@@ -369,6 +374,7 @@ public Response handle(OperationType op, String metalake, String parent, Excepti
   }
 
   private static class FilesetExceptionHandler extends BaseExceptionHandler {
+
     private static final ExceptionHandler INSTANCE = new FilesetExceptionHandler();
 
     private static String getFilesetErrorMsg(
@@ -520,6 +526,7 @@ public Response handle(OperationType op, String role, String metalake, Exception
   }
 
   private static class TopicExceptionHandler extends BaseExceptionHandler {
+
     private static final ExceptionHandler INSTANCE = new TopicExceptionHandler();
 
     private static String getTopicErrorMsg(
@@ -558,6 +565,7 @@ public Response handle(OperationType op, String topic, String schema, Exception
 
   private static class UserPermissionOperationExceptionHandler
       extends BasePermissionExceptionHandler {
+
     private static final ExceptionHandler INSTANCE = new UserPermissionOperationExceptionHandler();
 
     @Override
@@ -622,6 +630,35 @@ public Response handle(OperationType op, String roles, String parent, Exception
     }
   }
 
+  private static class CredentialExceptionHandler extends BaseExceptionHandler {
+
+    private static final ExceptionHandler INSTANCE = new CredentialExceptionHandler();
+
+    private static String getCredentialErrorMsg(String parent, String reason) {
+      return String.format(
+          "Failed to get credentials under object [%s], reason [%s]", parent, reason);
+    }
+
+    @Override
+    public Response handle(OperationType op, String credential, String parent, Exception e) {
+      String errorMsg = getCredentialErrorMsg(parent, getErrorMsg(e));
+      LOG.warn(errorMsg, e);
+
+      if (e instanceof IllegalArgumentException) {
+        return Utils.illegalArguments(errorMsg, e);
+
+      } else if (e instanceof NotFoundException) {
+        return Utils.notFound(errorMsg, e);
+
+      } else if (e instanceof NotInUseException) {
+        return Utils.notInUse(errorMsg, e);
+
+      } else {
+        return super.handle(op, credential, parent, e);
+      }
+    }
+  }
+
   private static class TagExceptionHandler extends BaseExceptionHandler {
 
     private static final ExceptionHandler INSTANCE = new TagExceptionHandler();
@@ -661,6 +698,7 @@ public Response handle(OperationType op, String tag, String parent, Exception e)
   }
 
   private static class OwnerExceptionHandler extends BaseExceptionHandler {
+
     private static final ExceptionHandler INSTANCE = new OwnerExceptionHandler();
 
     private static String getOwnerErrorMsg(

server/src/main/java/org/apache/gravitino/server/web/rest/MetadataObjectCredentialOperations.java

@@ -0,0 +1,100 @@
+/*
+ *  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.server.web.rest;
+
+import com.codahale.metrics.annotation.ResponseMetered;
+import com.codahale.metrics.annotation.Timed;
+import java.util.List;
+import java.util.Locale;
+import javax.inject.Inject;
+import javax.servlet.http.HttpServletRequest;
+import javax.ws.rs.GET;
+import javax.ws.rs.Path;
+import javax.ws.rs.PathParam;
+import javax.ws.rs.Produces;
+import javax.ws.rs.core.Context;
+import javax.ws.rs.core.Response;
+import org.apache.gravitino.MetadataObject;
+import org.apache.gravitino.MetadataObjects;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.catalog.CredentialManager;
+import org.apache.gravitino.credential.Credential;
+import org.apache.gravitino.dto.credential.CredentialDTO;
+import org.apache.gravitino.dto.responses.CredentialResponse;
+import org.apache.gravitino.dto.util.DTOConverters;
+import org.apache.gravitino.metrics.MetricNames;
+import org.apache.gravitino.server.web.Utils;
+import org.apache.gravitino.utils.MetadataObjectUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@Path("/metalakes/{metalake}/objects/{type}/{fullName}/credentials")
+public class MetadataObjectCredentialOperations {
+
+  private static final Logger LOG =
+      LoggerFactory.getLogger(MetadataObjectCredentialOperations.class);
+
+  private CredentialManager credentialManager;
+
+  @SuppressWarnings("unused")
+  @Context
+  private HttpServletRequest httpRequest;
+
+  @Inject
+  public MetadataObjectCredentialOperations(CredentialManager dispatcher) {
+    this.credentialManager = dispatcher;
+  }
+
+  @GET
+  @Produces("application/vnd.gravitino.v1+json")
+  @Timed(name = "get-credentials." + MetricNames.HTTP_PROCESS_DURATION, absolute = true)
+  @ResponseMetered(name = "get-credentials", absolute = true)
+  public Response getCredentials(
+      @PathParam("metalake") String metalake,
+      @PathParam("type") String type,
+      @PathParam("fullName") String fullName) {
+    LOG.info(
+        "Received get credentials request for object type: {}, full name: {} under metalake: {}",
+        type,
+        fullName,
+        metalake);
+
+    try {
+      return Utils.doAs(
+          httpRequest,
+          () -> {
+            MetadataObject object =
+                MetadataObjects.parse(
+                    fullName, MetadataObject.Type.valueOf(type.toUpperCase(Locale.ROOT)));
+
+            NameIdentifier identifier = MetadataObjectUtil.toEntityIdent(metalake, object);
+            List<Credential> credentials = credentialManager.getCredentials(identifier);
+            if (credentials == null) {
+              return Utils.ok(new CredentialResponse(new CredentialDTO[0]));
+            }
+            return Utils.ok(
+                new CredentialResponse(
+                    DTOConverters.toDTO(credentials.toArray(new Credential[credentials.size()]))));
+          });
+    } catch (Exception e) {
+      return ExceptionHandlers.handleCredentialException(OperationType.GET, fullName, e);
+    }
+  }
+}