[FLINK-37093][table] Fix catalog that failed validation due to no typ…

…e still exists in catalogStoreHolder
apache · Jan 17, 2025 · ab3a228 · ab3a228
1 parent d5ee298
commit ab3a228
Show file tree

Hide file tree

Showing 2 changed files with 14 additions and 3 deletions.
diff --git a/...ble/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/CatalogManager.java b/...ble/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-api-java/src/test/java/org/apache/flink/table/catalog/CatalogManagerTest.java b/...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());