[FLINK-37093][table] store catalog in catalogStoreHolder should after initCatalog and open when createCatalog in CatalogManager

Author: jiefei30

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

flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/catalog/CatalogManagerTest.java

@@ -546,6 +546,17 @@ 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'.");
 
         assertTrue(catalogManager.getCatalog("cat1").isPresent());
         assertTrue(catalogManager.getCatalog("cat2").isPresent());
@@ -589,6 +600,7 @@ void testCatalogStore() throws Exception {
         assertTrue(catalogManager.listCatalogs().contains("cat2"));
         assertTrue(catalogManager.listCatalogs().contains("cat3"));
         assertTrue(catalogManager.listCatalogs().contains("cat_comment"));
+        assertFalse(catalogManager.listCatalogs().contains("cat_no_type"));
 
         catalogManager.registerCatalog("cat4", new GenericInMemoryCatalog("cat4"));