[FLINK-19227][table] The catalog is still created after opening faile…

…d in catalog registering This closes apache#13414
alec-brooks · Sep 23, 2020 · 37f90c1 · 37f90c1
1 parent 452b043
commit 37f90c1
Show file tree

Hide file tree

Showing 2 changed files with 27 additions and 1 deletion.
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
@@ -187,8 +187,8 @@ public void registerCatalog(String catalogName, Catalog catalog) {
 			throw new CatalogException(format("Catalog %s already exists.", catalogName));
 		}
 
-		catalogs.put(catalogName, catalog);
 		catalog.open();
+		catalogs.put(catalogName, catalog);
 	}
 
 	/**

diff --git a/...ble-api-java/src/test/java/org/apache/flink/table/catalog/GenericInMemoryCatalogTest.java b/...ble-api-java/src/test/java/org/apache/flink/table/catalog/GenericInMemoryCatalogTest.java
@@ -18,6 +18,7 @@
 
 package org.apache.flink.table.catalog;
 
+import org.apache.flink.table.catalog.exceptions.CatalogException;
 import org.apache.flink.table.catalog.stats.CatalogColumnStatistics;
 import org.apache.flink.table.catalog.stats.CatalogColumnStatisticsDataBase;
 import org.apache.flink.table.catalog.stats.CatalogColumnStatisticsDataBinary;
@@ -30,14 +31,17 @@
 import org.apache.flink.table.catalog.stats.Date;
 import org.apache.flink.table.functions.TestGenericUDF;
 import org.apache.flink.table.functions.TestSimpleUDF;
+import org.apache.flink.table.utils.TableEnvironmentMock;
 
 import org.junit.BeforeClass;
 import org.junit.Test;
 
 import java.util.HashMap;
 import java.util.Map;
 
+import static org.hamcrest.CoreMatchers.equalTo;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
 
 /**
@@ -169,4 +173,26 @@ protected CatalogFunction createAnotherFunction() {
 	protected CatalogFunction createPythonFunction() {
 		return new CatalogFunctionImpl("test.func1", FunctionLanguage.PYTHON);
 	}
+
+	@Test
+	public void testRegisterCatalog() {
+		final TableEnvironmentMock tableEnv = TableEnvironmentMock.getStreamingInstance();
+		try {
+			tableEnv.registerCatalog(TEST_CATALOG_NAME, new MyCatalog(TEST_CATALOG_NAME));
+		} catch (CatalogException e) {
+		}
+		assertThat(tableEnv.getCatalog(TEST_CATALOG_NAME).isPresent(), equalTo(false));
+	}
+
+	class MyCatalog extends GenericInMemoryCatalog {
+
+		public MyCatalog(String name) {
+			super(name);
+		}
+
+		@Override
+		public void open() {
+			throw new CatalogException("open catalog failed.");
+		}
+	}
 }