From ab3a228370c51329dd7f8247e6a52632bf0834ff Mon Sep 17 00:00:00 2001
From: jiefei <jiefei30@126.com>
Date: Fri, 10 Jan 2025 16:01:21 +0800
Subject: [PATCH] [FLINK-37093][table] Fix catalog that failed validation due
 to no type still exists in catalogStoreHolder

---
 .../apache/flink/table/catalog/CatalogManager.java   |  5 ++---
 .../flink/table/catalog/CatalogManagerTest.java      | 12 ++++++++++++
 2 files changed, 14 insertions(+), 3 deletions(-)

diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/CatalogManager.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/CatalogManager.java
index f5efc9e854a36..b985e5248b561 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/CatalogManager.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/CatalogManager.java
@@ -328,13 +328,12 @@ public void createCatalog(
                 throw new CatalogException(format("Catalog %s already exists.", catalogName));
             }
         } else {
-            // Store the catalog in the catalog store
-            catalogStoreHolder.catalogStore().storeCatalog(catalogName, catalogDescriptor);
-
             // Initialize and store the catalog in memory
             Catalog catalog = initCatalog(catalogName, catalogDescriptor);
             catalog.open();
             catalogs.put(catalogName, catalog);
+            // Store the catalog in the catalog store
+            catalogStoreHolder.catalogStore().storeCatalog(catalogName, catalogDescriptor);
         }
     }
 
diff --git a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/catalog/CatalogManagerTest.java b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/catalog/CatalogManagerTest.java
index 4f9614d8ac713..c241633385043 100644
--- a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/catalog/CatalogManagerTest.java
+++ b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/catalog/CatalogManagerTest.java
@@ -546,7 +546,19 @@ void testCatalogStore() throws Exception {
                                         false))
                 .isInstanceOf(CatalogException.class)
                 .hasMessage("Catalog cat_comment already exists.");
+        assertThatThrownBy(
+                        () ->
+                                catalogManager.createCatalog(
+                                        "cat_no_type",
+                                        CatalogDescriptor.of(
+                                                "cat_no_type",
+                                                new Configuration(),
+                                                "catalog without type"),
+                                        false))
+                .isInstanceOf(ValidationException.class)
+                .hasMessageContaining("Unable to create catalog 'cat_no_type'.");
 
+        assertFalse(catalogManager.listCatalogs().contains("cat_no_type"));
         assertTrue(catalogManager.getCatalog("cat1").isPresent());
         assertTrue(catalogManager.getCatalog("cat2").isPresent());
         assertTrue(catalogManager.getCatalog("cat3").isPresent());