From d851f3d24a293273d5286f0f6d0cae255d74bd46 Mon Sep 17 00:00:00 2001 From: Dale Lane Date: Fri, 17 Jul 2026 21:33:56 +0100 Subject: [PATCH 1/3] [FLINK-40166][table] SQL query parsing fails if current catalog unreachable If the current catalog is unreachable, any SQL query fails to parse - even queries that make fully-qualified accesses to catalogs that are reachable. This is because we make a call to databaseExists in the current catalog as part of parsing the statement. This commit wraps this in a try..catch so it doesn't block the remainder of the parsing. Signed-off-by: Dale Lane --- .../flink/table/catalog/CatalogManager.java | 16 +++- .../table/catalog/CatalogManagerTest.java | 37 +++++++++ .../catalog/BrokenCurrentCatalogTest.java | 62 +++++++++++++++ .../UnreachableTestCatalogFactory.java | 76 +++++++++++++++++++ .../org.apache.flink.table.factories.Factory | 1 + 5 files changed, 191 insertions(+), 1 deletion(-) create mode 100644 flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/catalog/BrokenCurrentCatalogTest.java create mode 100644 flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/UnreachableTestCatalogFactory.java 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 ca31e17ebf4e48..f325ef88369e88 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 @@ -1114,7 +1114,21 @@ private boolean temporaryDatabaseExists(String catalogName, String databaseName) } private boolean permanentDatabaseExists(String catalogName, String databaseName) { - return getCatalog(catalogName).map(c -> c.databaseExists(databaseName)).orElse(false); + return getCatalog(catalogName) + .map( + c -> { + try { + return c.databaseExists(databaseName); + } catch (CatalogException e) { + LOG.debug( + "Unable to check whether database '{}' exists in catalog '{}'.", + databaseName, + catalogName, + e); + return false; + } + }) + .orElse(false); } /** 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 907d03dfcf8d1a..7b735d12490648 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 @@ -245,6 +245,43 @@ void testTableModificationListener() throws Exception { assertThat(dropTemporaryEvent.identifier().getObjectName()).isEqualTo("table2"); } + @Test + void testSchemaExistsSwallowsCatalogExceptionFromDatabaseExists() { + CatalogManager catalogManager = + CatalogManagerMocks.preparedCatalogManager() + .defaultCatalog("broken", new UnreachableCatalog("broken")) + .classLoader(CatalogManagerTest.class.getClassLoader()) + .config(new Configuration()) + .catalogStoreHolder( + CatalogStoreHolder.newBuilder() + .classloader(CatalogManagerTest.class.getClassLoader()) + .catalogStore(new GenericInMemoryCatalogStore()) + .config(new Configuration()) + .build()) + .build(); + assertThat(catalogManager.schemaExists("broken", "default")).isFalse(); + } + + /** + * A catalog whose {@link #databaseExists(String)} always fails, simulating a connectivity + * problem with an unreachable destination. + */ + private static class UnreachableCatalog extends GenericInMemoryCatalog { + UnreachableCatalog(String name) { + super(name, "default"); + } + + @Override + public boolean databaseExists(String databaseName) { + throw new CatalogException( + "Failed to connect for database '" + + databaseName + + "' of catalog '" + + getName() + + "'."); + } + } + @Test public void testDropCurrentDatabase() throws Exception { CatalogManager catalogManager = createCatalogManager(null); diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/catalog/BrokenCurrentCatalogTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/catalog/BrokenCurrentCatalogTest.java new file mode 100644 index 00000000000000..37fffe53509343 --- /dev/null +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/catalog/BrokenCurrentCatalogTest.java @@ -0,0 +1,62 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.table.planner.catalog; + +import org.apache.flink.configuration.Configuration; +import org.apache.flink.table.api.EnvironmentSettings; +import org.apache.flink.table.api.Table; +import org.apache.flink.table.api.TableEnvironment; +import org.apache.flink.table.catalog.CatalogDescriptor; +import org.apache.flink.table.planner.factories.UnreachableTestCatalogFactory; + +import org.junit.jupiter.api.Test; + +import static org.apache.flink.table.catalog.GenericInMemoryCatalogFactoryOptions.IDENTIFIER; +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Tests that a fully-qualified query against a healthy catalog can still be resolved even when the + * current catalog is unreachable. + */ +class BrokenCurrentCatalogTest { + + @Test + void testFullyQualifiedQueryWhileCurrentCatalogIsBroken() throws Exception { + TableEnvironment tEnv = + TableEnvironment.create( + EnvironmentSettings.newInstance().inStreamingMode().build()); + + Configuration healthyOptions = new Configuration(); + healthyOptions.setString("type", IDENTIFIER); + healthyOptions.setString("default-database", "default"); + tEnv.createCatalog("healthy", CatalogDescriptor.of("healthy", healthyOptions)); + tEnv.useCatalog("healthy"); + tEnv.executeSql( + "CREATE VIEW `healthy`.`default`.`v` AS SELECT * FROM (VALUES (1), (2), (3)) AS t(id)"); + + Configuration brokenOptions = new Configuration(); + brokenOptions.setString("type", UnreachableTestCatalogFactory.IDENTIFIER); + tEnv.createCatalog("broken", CatalogDescriptor.of("broken", brokenOptions)); + tEnv.useCatalog("broken"); + + Table table = tEnv.sqlQuery("SELECT * FROM `healthy`.`default`.`v`"); + + assertThat(table.getResolvedSchema().getColumnNames()).containsExactly("id"); + } +} diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/UnreachableTestCatalogFactory.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/UnreachableTestCatalogFactory.java new file mode 100644 index 00000000000000..d7e38f81f7bb10 --- /dev/null +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/UnreachableTestCatalogFactory.java @@ -0,0 +1,76 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.table.planner.factories; + +import org.apache.flink.configuration.ConfigOption; +import org.apache.flink.table.catalog.Catalog; +import org.apache.flink.table.catalog.GenericInMemoryCatalog; +import org.apache.flink.table.catalog.exceptions.CatalogException; +import org.apache.flink.table.factories.CatalogFactory; +import org.apache.flink.table.factories.FactoryUtil; + +import java.util.Collections; +import java.util.Set; + +/** + * Test catalog factory that creates a catalog whose {@link Catalog#databaseExists(String)} always + * fails, simulating a catalog that cannot be reached (e.g. a connectivity failure). + */ +public class UnreachableTestCatalogFactory implements CatalogFactory { + + public static final String IDENTIFIER = "test-unreachable-catalog"; + + @Override + public String factoryIdentifier() { + return IDENTIFIER; + } + + @Override + public Set> requiredOptions() { + return Collections.emptySet(); + } + + @Override + public Set> optionalOptions() { + return Collections.emptySet(); + } + + @Override + public Catalog createCatalog(Context context) { + FactoryUtil.createCatalogFactoryHelper(this, context).validate(); + return new UnreachableCatalog(context.getName()); + } + + /** A catalog whose {@link #databaseExists(String)} always fails. */ + public static class UnreachableCatalog extends GenericInMemoryCatalog { + public UnreachableCatalog(String name) { + super(name, "default"); + } + + @Override + public boolean databaseExists(String databaseName) { + throw new CatalogException( + "Failed to connect to Kafka cluster for database '" + + databaseName + + "' of catalog '" + + getName() + + "'."); + } + } +} diff --git a/flink-table/flink-table-planner/src/test/resources/META-INF/services/org.apache.flink.table.factories.Factory b/flink-table/flink-table-planner/src/test/resources/META-INF/services/org.apache.flink.table.factories.Factory index 5f055635a8ae7a..bfbca086a6fcee 100644 --- a/flink-table/flink-table-planner/src/test/resources/META-INF/services/org.apache.flink.table.factories.Factory +++ b/flink-table/flink-table-planner/src/test/resources/META-INF/services/org.apache.flink.table.factories.Factory @@ -25,3 +25,4 @@ org.apache.flink.table.planner.utils.TestSimpleDynamicTableSourceFactory org.apache.flink.table.planner.factories.TestValuesModelFactory org.apache.flink.table.planner.factories.PlainTestCatalogFactory org.apache.flink.table.planner.factories.TimeoutAsyncLookupTableFactory +org.apache.flink.table.planner.factories.UnreachableTestCatalogFactory From 4fc21e2197b3222af43beb9a2ed3c7779db66d8f Mon Sep 17 00:00:00 2001 From: Dale Lane Date: Wed, 22 Jul 2026 22:59:48 +0100 Subject: [PATCH 2/3] fix: address code review comments Signed-off-by: Dale Lane --- .../flink/table/catalog/CatalogManager.java | 2 +- .../catalog/BrokenCurrentCatalogTest.java | 31 +++++--- .../UnreachableTestCatalogFactory.java | 76 ------------------- .../org.apache.flink.table.factories.Factory | 1 - 4 files changed, 21 insertions(+), 89 deletions(-) delete mode 100644 flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/UnreachableTestCatalogFactory.java 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 f325ef88369e88..fa9db8cb50dbda 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 @@ -1120,7 +1120,7 @@ private boolean permanentDatabaseExists(String catalogName, String databaseName) try { return c.databaseExists(databaseName); } catch (CatalogException e) { - LOG.debug( + LOG.warn( "Unable to check whether database '{}' exists in catalog '{}'.", databaseName, catalogName, diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/catalog/BrokenCurrentCatalogTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/catalog/BrokenCurrentCatalogTest.java index 37fffe53509343..8641f761b2dede 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/catalog/BrokenCurrentCatalogTest.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/catalog/BrokenCurrentCatalogTest.java @@ -18,16 +18,14 @@ package org.apache.flink.table.planner.catalog; -import org.apache.flink.configuration.Configuration; import org.apache.flink.table.api.EnvironmentSettings; import org.apache.flink.table.api.Table; import org.apache.flink.table.api.TableEnvironment; -import org.apache.flink.table.catalog.CatalogDescriptor; -import org.apache.flink.table.planner.factories.UnreachableTestCatalogFactory; +import org.apache.flink.table.catalog.GenericInMemoryCatalog; +import org.apache.flink.table.catalog.exceptions.CatalogException; import org.junit.jupiter.api.Test; -import static org.apache.flink.table.catalog.GenericInMemoryCatalogFactoryOptions.IDENTIFIER; import static org.assertj.core.api.Assertions.assertThat; /** @@ -42,21 +40,32 @@ void testFullyQualifiedQueryWhileCurrentCatalogIsBroken() throws Exception { TableEnvironment.create( EnvironmentSettings.newInstance().inStreamingMode().build()); - Configuration healthyOptions = new Configuration(); - healthyOptions.setString("type", IDENTIFIER); - healthyOptions.setString("default-database", "default"); - tEnv.createCatalog("healthy", CatalogDescriptor.of("healthy", healthyOptions)); + tEnv.registerCatalog("healthy", new GenericInMemoryCatalog("healthy", "default")); tEnv.useCatalog("healthy"); tEnv.executeSql( "CREATE VIEW `healthy`.`default`.`v` AS SELECT * FROM (VALUES (1), (2), (3)) AS t(id)"); - Configuration brokenOptions = new Configuration(); - brokenOptions.setString("type", UnreachableTestCatalogFactory.IDENTIFIER); - tEnv.createCatalog("broken", CatalogDescriptor.of("broken", brokenOptions)); + tEnv.registerCatalog("broken", new UnreachableCatalog("broken")); tEnv.useCatalog("broken"); Table table = tEnv.sqlQuery("SELECT * FROM `healthy`.`default`.`v`"); assertThat(table.getResolvedSchema().getColumnNames()).containsExactly("id"); } + + private static class UnreachableCatalog extends GenericInMemoryCatalog { + UnreachableCatalog(String name) { + super(name, "default"); + } + + @Override + public boolean databaseExists(String databaseName) { + throw new CatalogException( + "Failed to connect to database '" + + databaseName + + "' of catalog '" + + getName() + + "'."); + } + } } diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/UnreachableTestCatalogFactory.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/UnreachableTestCatalogFactory.java deleted file mode 100644 index d7e38f81f7bb10..00000000000000 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/UnreachableTestCatalogFactory.java +++ /dev/null @@ -1,76 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.apache.flink.table.planner.factories; - -import org.apache.flink.configuration.ConfigOption; -import org.apache.flink.table.catalog.Catalog; -import org.apache.flink.table.catalog.GenericInMemoryCatalog; -import org.apache.flink.table.catalog.exceptions.CatalogException; -import org.apache.flink.table.factories.CatalogFactory; -import org.apache.flink.table.factories.FactoryUtil; - -import java.util.Collections; -import java.util.Set; - -/** - * Test catalog factory that creates a catalog whose {@link Catalog#databaseExists(String)} always - * fails, simulating a catalog that cannot be reached (e.g. a connectivity failure). - */ -public class UnreachableTestCatalogFactory implements CatalogFactory { - - public static final String IDENTIFIER = "test-unreachable-catalog"; - - @Override - public String factoryIdentifier() { - return IDENTIFIER; - } - - @Override - public Set> requiredOptions() { - return Collections.emptySet(); - } - - @Override - public Set> optionalOptions() { - return Collections.emptySet(); - } - - @Override - public Catalog createCatalog(Context context) { - FactoryUtil.createCatalogFactoryHelper(this, context).validate(); - return new UnreachableCatalog(context.getName()); - } - - /** A catalog whose {@link #databaseExists(String)} always fails. */ - public static class UnreachableCatalog extends GenericInMemoryCatalog { - public UnreachableCatalog(String name) { - super(name, "default"); - } - - @Override - public boolean databaseExists(String databaseName) { - throw new CatalogException( - "Failed to connect to Kafka cluster for database '" - + databaseName - + "' of catalog '" - + getName() - + "'."); - } - } -} diff --git a/flink-table/flink-table-planner/src/test/resources/META-INF/services/org.apache.flink.table.factories.Factory b/flink-table/flink-table-planner/src/test/resources/META-INF/services/org.apache.flink.table.factories.Factory index bfbca086a6fcee..5f055635a8ae7a 100644 --- a/flink-table/flink-table-planner/src/test/resources/META-INF/services/org.apache.flink.table.factories.Factory +++ b/flink-table/flink-table-planner/src/test/resources/META-INF/services/org.apache.flink.table.factories.Factory @@ -25,4 +25,3 @@ org.apache.flink.table.planner.utils.TestSimpleDynamicTableSourceFactory org.apache.flink.table.planner.factories.TestValuesModelFactory org.apache.flink.table.planner.factories.PlainTestCatalogFactory org.apache.flink.table.planner.factories.TimeoutAsyncLookupTableFactory -org.apache.flink.table.planner.factories.UnreachableTestCatalogFactory From 3827ebcacad71117b5ea9398c213dfbd1fcc32a5 Mon Sep 17 00:00:00 2001 From: Dale Lane Date: Thu, 23 Jul 2026 09:18:22 +0100 Subject: [PATCH 3/3] fix: typo in message Signed-off-by: Dale Lane --- .../java/org/apache/flink/table/catalog/CatalogManagerTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 7b735d12490648..51774ffb6f6b17 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 @@ -274,7 +274,7 @@ private static class UnreachableCatalog extends GenericInMemoryCatalog { @Override public boolean databaseExists(String databaseName) { throw new CatalogException( - "Failed to connect for database '" + "Failed to connect to database '" + databaseName + "' of catalog '" + getName()