diff --git a/build.gradle.kts b/build.gradle.kts
index 1309e17b198..2f4ba7fca49 100644
--- a/build.gradle.kts
+++ b/build.gradle.kts
@@ -505,6 +505,7 @@ tasks.rat {
     "ROADMAP.md",
     "clients/client-python/.pytest_cache/*",
     "clients/client-python/.venv/*",
+    "clients/client-python/venv/*",
     "clients/client-python/apache_gravitino.egg-info/*",
     "clients/client-python/gravitino/utils/http_client.py",
     "clients/client-python/tests/unittests/htmlcov/*",
diff --git a/core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java b/core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java
index f351a3271fe..9dc1d0ed277 100644
--- a/core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java
+++ b/core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java
@@ -366,7 +366,7 @@ public Catalog createCatalog(
                     .build())
             .build();
 
-    boolean createSuccess = false;
+    boolean needClean = true;
     try {
       NameIdentifier metalakeIdent = NameIdentifier.of(ident.namespace().levels());
       if (!store.exists(metalakeIdent, EntityType.METALAKE)) {
@@ -376,13 +376,18 @@ public Catalog createCatalog(
 
       store.put(e, false /* overwrite */);
       CatalogWrapper wrapper = catalogCache.get(ident, id -> createCatalogWrapper(e));
-      createSuccess = true;
+
+      needClean = false;
       return wrapper.catalog;
+
     } catch (EntityAlreadyExistsException e1) {
+      needClean = false;
       LOG.warn("Catalog {} already exists", ident, e1);
       throw new CatalogAlreadyExistsException("Catalog %s already exists", ident);
+
     } catch (IllegalArgumentException | NoSuchMetalakeException e2) {
       throw e2;
+
     } catch (Exception e3) {
       catalogCache.invalidate(ident);
       LOG.error("Failed to create catalog {}", ident, e3);
@@ -390,8 +395,11 @@ public Catalog createCatalog(
         throw (RuntimeException) e3;
       }
       throw new RuntimeException(e3);
+
     } finally {
-      if (!createSuccess) {
+      if (needClean) {
+        // since we put the catalog entity into the store but failed to create the catalog instance,
+        // we need to clean up the entity stored.
         try {
           store.delete(ident, EntityType.CATALOG, true);
         } catch (IOException e4) {
diff --git a/core/src/test/java/org/apache/gravitino/catalog/TestCatalogManager.java b/core/src/test/java/org/apache/gravitino/catalog/TestCatalogManager.java
index 7f6c52811a3..730f94fb9b8 100644
--- a/core/src/test/java/org/apache/gravitino/catalog/TestCatalogManager.java
+++ b/core/src/test/java/org/apache/gravitino/catalog/TestCatalogManager.java
@@ -269,10 +269,10 @@ public void testCreateCatalog() {
 
     // test before creation
     Assertions.assertThrows(
-        NoSuchMetalakeException.class,
+        CatalogAlreadyExistsException.class,
         () ->
             catalogManager.testConnection(
-                ident2, Catalog.Type.RELATIONAL, provider, "comment", props));
+                ident, Catalog.Type.RELATIONAL, provider, "comment", props));
 
     // Test create with duplicated name
     Throwable exception2 =
@@ -309,7 +309,8 @@ public void testCreateCatalog() {
                 catalogManager.createCatalog(
                     failedIdent, Catalog.Type.RELATIONAL, provider, "comment", props));
     Assertions.assertTrue(
-        exception4.getMessage().contains("Properties are reserved and cannot be set"));
+        exception4.getMessage().contains("Properties are reserved and cannot be set"),
+        exception4.getMessage());
     Assertions.assertNull(catalogManager.catalogCache.getIfPresent(failedIdent));
   }
 
diff --git a/integration-test/src/test/java/org/apache/gravitino/integration/test/client/CatalogIT.java b/integration-test/src/test/java/org/apache/gravitino/integration/test/client/CatalogIT.java
index 57dcd66cae6..fca39edfc9f 100644
--- a/integration-test/src/test/java/org/apache/gravitino/integration/test/client/CatalogIT.java
+++ b/integration-test/src/test/java/org/apache/gravitino/integration/test/client/CatalogIT.java
@@ -28,6 +28,7 @@
 import org.apache.gravitino.Catalog;
 import org.apache.gravitino.CatalogChange;
 import org.apache.gravitino.client.GravitinoMetalake;
+import org.apache.gravitino.exceptions.CatalogAlreadyExistsException;
 import org.apache.gravitino.integration.test.container.ContainerSuite;
 import org.apache.gravitino.integration.test.container.HiveContainer;
 import org.apache.gravitino.integration.test.util.AbstractIT;
@@ -125,6 +126,13 @@ public void testDropCatalog() {
     Assertions.assertTrue(metalake.catalogExists(catalogName));
     Assertions.assertEquals(catalogName, catalog.name());
 
+    Assertions.assertThrows(
+        CatalogAlreadyExistsException.class,
+        () ->
+            metalake.createCatalog(
+                catalogName, Catalog.Type.RELATIONAL, "hive", "catalog comment", properties));
+    Assertions.assertTrue(metalake.catalogExists(catalogName));
+
     Assertions.assertTrue(metalake.dropCatalog(catalogName), "catalog should be dropped");
     Assertions.assertFalse(metalake.dropCatalog(catalogName), "catalog should be non-existent");
   }