[#6716] improvement(authz): Delete catalogs if failing to execute post hook actions

authorizations/authorization-ranger/src/main/java/org/apache/gravitino/authorization/ranger/RangerAuthorizationPlugin.java

@@ -103,10 +103,7 @@ protected RangerAuthorizationPlugin(String metalake, Map<String, String> config)
     rangerServiceName = config.get(RangerAuthorizationProperties.RANGER_SERVICE_NAME);
     rangerClient = new RangerClientExtension(rangerUrl, authType, rangerAdminName, password);
 
-    if (Boolean.parseBoolean(
-        config.get(RangerAuthorizationProperties.RANGER_SERVICE_CREATE_IF_ABSENT))) {
-      createRangerServiceIfNecessary(config, rangerServiceName);
-    }
+    createRangerServiceIfNecessary(config, rangerServiceName);
 
     rangerHelper =
         new RangerHelper(
@@ -786,7 +783,9 @@ private void createRangerServiceIfNecessary(Map<String, String> config, String s
     try {
       rangerClient.getService(serviceName);
     } catch (RangerServiceException rse) {
-      if (rse.getStatus().equals(ClientResponse.Status.NOT_FOUND)) {
+      if (Boolean.parseBoolean(
+              config.get(RangerAuthorizationProperties.RANGER_SERVICE_CREATE_IF_ABSENT))
+          && ClientResponse.Status.NOT_FOUND.equals(rse.getStatus())) {
         try {
           RangerService rangerService = new RangerService();
           rangerService.setType(getServiceType());

authorizations/authorization-ranger/src/test/java/org/apache/gravitino/authorization/ranger/integration/test/RangerHiveE2EIT.java

@@ -237,6 +237,45 @@ public void createCatalog() {
     } catch (Exception e) {
       throw new RuntimeException(e);
     }
+
+    // Test to create a catalog with wrong properties which lacks Ranger service name,
+    // It will throw RuntimeException with message that Ranger service name is required.
+    Map<String, String> wrongProperties =
+        ImmutableMap.of(
+            HiveConstants.METASTORE_URIS,
+            HIVE_METASTORE_URIS,
+            IMPERSONATION_ENABLE,
+            "true",
+            AUTHORIZATION_PROVIDER,
+            "ranger",
+            RangerAuthorizationProperties.RANGER_SERVICE_TYPE,
+            "HadoopSQL",
+            RangerAuthorizationProperties.RANGER_ADMIN_URL,
+            RangerITEnv.RANGER_ADMIN_URL,
+            RangerAuthorizationProperties.RANGER_AUTH_TYPE,
+            RangerContainer.authType,
+            RangerAuthorizationProperties.RANGER_USERNAME,
+            RangerContainer.rangerUserName,
+            RangerAuthorizationProperties.RANGER_PASSWORD,
+            RangerContainer.rangerPassword,
+            RangerAuthorizationProperties.RANGER_SERVICE_CREATE_IF_ABSENT,
+            "true");
+
+    int catalogSize = metalake.listCatalogs().length;
+    Exception exception =
+        Assertions.assertThrows(
+            RuntimeException.class,
+            () ->
+                metalake.createCatalog(
+                    "wrongTestProperties",
+                    Catalog.Type.RELATIONAL,
+                    provider,
+                    "comment",
+                    wrongProperties));
+    Assertions.assertTrue(
+        exception.getMessage().contains("authorization.ranger.service.name is required"));
+
+    Assertions.assertEquals(catalogSize, metalake.listCatalogs().length);
   }
 
   protected void checkTableAllPrivilegesExceptForCreating() {

core/src/main/java/org/apache/gravitino/hook/CatalogHookDispatcher.java

@@ -40,13 +40,16 @@
 import org.apache.gravitino.exceptions.NonEmptyEntityException;
 import org.apache.gravitino.utils.NameIdentifierUtil;
 import org.apache.gravitino.utils.PrincipalUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * {@code CatalogHookDispatcher} is a decorator for {@link CatalogDispatcher} that not only
  * delegates catalog operations to the underlying catalog dispatcher but also executes some hook
  * operations before or after the underlying operations.
  */
 public class CatalogHookDispatcher implements CatalogDispatcher {
+  private static final Logger LOG = LoggerFactory.getLogger(CatalogHookDispatcher.class);
   private final CatalogDispatcher dispatcher;
 
   public CatalogHookDispatcher(CatalogDispatcher dispatcher) {
@@ -82,21 +85,26 @@ public Catalog createCatalog(
 
     Catalog catalog = dispatcher.createCatalog(ident, type, provider, comment, properties);
 
-    // Set the creator as the owner of the catalog.
-    OwnerManager ownerManager = GravitinoEnv.getInstance().ownerManager();
-    if (ownerManager != null) {
-      ownerManager.setOwner(
-          ident.namespace().level(0),
-          NameIdentifierUtil.toMetadataObject(ident, Entity.EntityType.CATALOG),
-          PrincipalUtils.getCurrentUserName(),
-          Owner.Type.USER);
-    }
+    try {
+      // Set the creator as the owner of the catalog.
+      OwnerManager ownerManager = GravitinoEnv.getInstance().ownerManager();
+      if (ownerManager != null) {
+        ownerManager.setOwner(
+            ident.namespace().level(0),
+            NameIdentifierUtil.toMetadataObject(ident, Entity.EntityType.CATALOG),
+            PrincipalUtils.getCurrentUserName(),
+            Owner.Type.USER);
+      }
 
-    // Apply the metalake securable object privileges to authorization plugin
-    FutureGrantManager futureGrantManager = GravitinoEnv.getInstance().futureGrantManager();
-    if (futureGrantManager != null && catalog instanceof BaseCatalog) {
-      futureGrantManager.grantNewlyCreatedCatalog(
-          ident.namespace().level(0), (BaseCatalog) catalog);
+      // Apply the metalake securable object privileges to authorization plugin
+      FutureGrantManager futureGrantManager = GravitinoEnv.getInstance().futureGrantManager();
+      if (futureGrantManager != null && catalog instanceof BaseCatalog) {
+        futureGrantManager.grantNewlyCreatedCatalog(
+            ident.namespace().level(0), (BaseCatalog) catalog);
+      }
+    } catch (Exception e) {
+      LOG.warn("Fail to execute the post hook operations, rollback the catalog " + ident, e);
+      dispatcher.dropCatalog(ident, true);
     }
 
     return catalog;