This is an automated email from the ASF dual-hosted git repository.
diqiu50 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new e783912f21 [#13213] improvement(flink-connector): Recognize catalog
types from external BaseCatalogFactory jars (#13214)
e783912f21 is described below
commit e783912f21a3ab5676f89651285ddfcbdf201891
Author: Yuhui <[email protected]>
AuthorDate: Thu Sep 17 10:03:06 2026 +0800
[#13213] improvement(flink-connector): Recognize catalog types from
external BaseCatalogFactory jars (#13214)
### What changes were proposed in this pull request?
`FactoryUtils.isGravitinoManagedCatalogType` now also accepts any type
declared by a `BaseCatalogFactory` found through `ServiceLoader`, so a
catalog jar outside the connector can contribute catalog types. Factory
entries that fail to load are skipped.
### Why are the changes needed?
External catalog implementations were discoverable by Flink but not
recognized as Gravitino-managed by the connector.
Fix: #13213
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Added `TestFactoryUtils`.
---
flink-connector/flink-common/build.gradle.kts | 2 +
.../flink/connector/utils/FactoryUtils.java | 41 +++++++-
.../utils/FakeExternalCatalogFactory.java | 73 +++++++++++++
.../flink/connector/utils/TestFactoryUtils.java | 117 +++++++++++++++++++++
.../org.apache.flink.table.factories.Factory | 20 ++++
5 files changed, 252 insertions(+), 1 deletion(-)
diff --git a/flink-connector/flink-common/build.gradle.kts
b/flink-connector/flink-common/build.gradle.kts
index 6e121d52a7..9938408fad 100644
--- a/flink-connector/flink-common/build.gradle.kts
+++ b/flink-connector/flink-common/build.gradle.kts
@@ -208,6 +208,8 @@ val testJar by tasks.registering(Jar::class) {
archiveClassifier.set("tests")
archiveBaseName.set(artifactName)
from(sourceSets["test"].output)
+ // The SPI fixture is only for this module's tests; keep it off the version
modules' classpath.
+ exclude("META-INF/services/**")
}
configurations {
diff --git
a/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/utils/FactoryUtils.java
b/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/utils/FactoryUtils.java
index 273da3f4ac..fc6fc9fb32 100644
---
a/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/utils/FactoryUtils.java
+++
b/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/utils/FactoryUtils.java
@@ -21,13 +21,18 @@ package org.apache.gravitino.flink.connector.utils;
import static
org.apache.flink.table.factories.FactoryUtil.validateFactoryOptions;
import static
org.apache.flink.table.factories.FactoryUtil.validateWatermarkOptions;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableSet;
import java.util.HashSet;
+import java.util.Iterator;
+import java.util.ServiceConfigurationError;
+import java.util.ServiceLoader;
import java.util.Set;
import java.util.stream.Collectors;
import org.apache.flink.table.factories.CatalogFactory;
import org.apache.flink.table.factories.Factory;
import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.gravitino.flink.connector.catalog.BaseCatalogFactory;
import
org.apache.gravitino.flink.connector.hive.GravitinoHiveCatalogFactoryOptions;
import
org.apache.gravitino.flink.connector.iceberg.GravitinoIcebergCatalogFactoryOptions;
import
org.apache.gravitino.flink.connector.jdbc.GravitinoJdbcCatalogFactoryOptions;
@@ -51,7 +56,41 @@ public class FactoryUtils {
GravitinoPaimonCatalogFactoryOptions.IDENTIFIER);
public static boolean isGravitinoManagedCatalogType(String type) {
- return GRAVITINO_CATALOG_TYPES.contains(type);
+ return GRAVITINO_CATALOG_TYPES.contains(type) ||
isProvidedByCatalogFactory(type);
+ }
+
+ /**
+ * Whether a {@link BaseCatalogFactory} on the classpath declares the type.
This lets a jar
+ * outside the connector contribute catalog types without registering them
here.
+ */
+ private static boolean isProvidedByCatalogFactory(String type) {
+ if (type == null) {
+ return false;
+ }
+ return
isProvidedByCatalogFactory(ServiceLoader.load(Factory.class).iterator(), type);
+ }
+
+ @VisibleForTesting
+ static boolean isProvidedByCatalogFactory(Iterator<Factory> factories,
String type) {
+ while (true) {
+ try {
+ if (!factories.hasNext()) {
+ return false;
+ }
+ Factory factory = factories.next();
+ if (factory instanceof BaseCatalogFactory &&
type.equals(factory.factoryIdentifier())) {
+ return true;
+ }
+ } catch (ServiceConfigurationError | LinkageError | RuntimeException e) {
+ // A factory that cannot be loaded or fails to report its identifier
cannot be the one
+ // asked for.
+ LOG.debug(
+ "Skip a {} entry that cannot be loaded while resolving catalog
type '{}'.",
+ Factory.class.getName(),
+ type,
+ e);
+ }
+ }
}
/**
diff --git
a/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/utils/FakeExternalCatalogFactory.java
b/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/utils/FakeExternalCatalogFactory.java
new file mode 100644
index 0000000000..664ceeda5d
--- /dev/null
+++
b/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/utils/FakeExternalCatalogFactory.java
@@ -0,0 +1,73 @@
+/*
+ * 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.gravitino.flink.connector.utils;
+
+import java.util.Collections;
+import java.util.Set;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.gravitino.flink.connector.CatalogPropertiesConverter;
+import org.apache.gravitino.flink.connector.PartitionConverter;
+import org.apache.gravitino.flink.connector.catalog.BaseCatalogFactory;
+
+/** A catalog factory registered only on the test classpath, standing in for
an external jar. */
+public class FakeExternalCatalogFactory implements BaseCatalogFactory {
+
+ public static final String IDENTIFIER = "gravitino-fake-external";
+
+ @Override
+ public String factoryIdentifier() {
+ return IDENTIFIER;
+ }
+
+ @Override
+ public Set<ConfigOption<?>> requiredOptions() {
+ return Collections.emptySet();
+ }
+
+ @Override
+ public Set<ConfigOption<?>> optionalOptions() {
+ return Collections.emptySet();
+ }
+
+ @Override
+ public Catalog createCatalog(Context context) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public String gravitinoCatalogProvider() {
+ return "fake-external";
+ }
+
+ @Override
+ public org.apache.gravitino.Catalog.Type gravitinoCatalogType() {
+ return org.apache.gravitino.Catalog.Type.RELATIONAL;
+ }
+
+ @Override
+ public CatalogPropertiesConverter catalogPropertiesConverter() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public PartitionConverter partitionConverter() {
+ throw new UnsupportedOperationException();
+ }
+}
diff --git
a/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/utils/TestFactoryUtils.java
b/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/utils/TestFactoryUtils.java
new file mode 100644
index 0000000000..3bcea53ddb
--- /dev/null
+++
b/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/utils/TestFactoryUtils.java
@@ -0,0 +1,117 @@
+/*
+ * 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.gravitino.flink.connector.utils;
+
+import com.google.common.collect.ImmutableList;
+import java.util.Iterator;
+import java.util.ServiceConfigurationError;
+import org.apache.flink.table.factories.Factory;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class TestFactoryUtils {
+
+ @Test
+ void testBuiltInTypesAreGravitinoManaged() {
+
Assertions.assertTrue(FactoryUtils.isGravitinoManagedCatalogType("gravitino-hive"));
+
Assertions.assertTrue(FactoryUtils.isGravitinoManagedCatalogType("gravitino-jdbc-mysql"));
+ }
+
+ @Test
+ void testExternalCatalogFactoryTypeIsGravitinoManaged() {
+ // Discovered through META-INF/services on the test classpath. Matching
follows Flink's own
+ // factory discovery and is case-sensitive.
+ Assertions.assertTrue(
+
FactoryUtils.isGravitinoManagedCatalogType(FakeExternalCatalogFactory.IDENTIFIER));
+ Assertions.assertFalse(
+ FactoryUtils.isGravitinoManagedCatalogType(
+ FakeExternalCatalogFactory.IDENTIFIER.toUpperCase()));
+ }
+
+ @Test
+ void testTypeWithoutCatalogFactoryIsNotGravitinoManaged() {
+ // No jar on this classpath registers a factory for these types.
+
Assertions.assertFalse(FactoryUtils.isGravitinoManagedCatalogType("gravitino-jdbc-oracle"));
+
Assertions.assertFalse(FactoryUtils.isGravitinoManagedCatalogType("gravitino-jdbc-custom"));
+ Assertions.assertFalse(FactoryUtils.isGravitinoManagedCatalogType(null));
+ }
+
+ @Test
+ void testFlinkNativeCatalogTypesAreNotGravitinoManaged() {
+ // These factories are on the test classpath but are not
BaseCatalogFactory.
+
Assertions.assertFalse(FactoryUtils.isGravitinoManagedCatalogType("generic_in_memory"));
+ Assertions.assertFalse(FactoryUtils.isGravitinoManagedCatalogType("hive"));
+ }
+
+ @Test
+ void testBrokenServiceEntriesAreSkipped() {
+ Iterator<Factory> factories =
+ new Iterator<Factory>() {
+ private final Iterator<Object> steps =
+ ImmutableList.of(
+ new ServiceConfigurationError("missing provider"),
+ new NoClassDefFoundError("missing/Dependency"),
+ new IllegalStateException("factory init failed"),
+ new FakeExternalCatalogFactory())
+ .iterator();
+
+ @Override
+ public boolean hasNext() {
+ return steps.hasNext();
+ }
+
+ @Override
+ public Factory next() {
+ Object step = steps.next();
+ if (step instanceof Error) {
+ throw (Error) step;
+ }
+ if (step instanceof RuntimeException) {
+ throw (RuntimeException) step;
+ }
+ return (Factory) step;
+ }
+ };
+ Assertions.assertTrue(
+ FactoryUtils.isProvidedByCatalogFactory(factories,
FakeExternalCatalogFactory.IDENTIFIER));
+ }
+
+ @Test
+ void testBrokenServiceEntryOnHasNextIsSkipped() {
+ Iterator<Factory> factories =
+ new Iterator<Factory>() {
+ private boolean failed = false;
+
+ @Override
+ public boolean hasNext() {
+ if (!failed) {
+ failed = true;
+ throw new ServiceConfigurationError("unreadable service file");
+ }
+ return false;
+ }
+
+ @Override
+ public Factory next() {
+ throw new IllegalStateException();
+ }
+ };
+ Assertions.assertFalse(FactoryUtils.isProvidedByCatalogFactory(factories,
"any"));
+ }
+}
diff --git
a/flink-connector/flink-common/src/test/resources/META-INF/services/org.apache.flink.table.factories.Factory
b/flink-connector/flink-common/src/test/resources/META-INF/services/org.apache.flink.table.factories.Factory
new file mode 100644
index 0000000000..d208e6d87f
--- /dev/null
+++
b/flink-connector/flink-common/src/test/resources/META-INF/services/org.apache.flink.table.factories.Factory
@@ -0,0 +1,20 @@
+#
+# 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.
+#
+
+org.apache.gravitino.flink.connector.utils.FakeExternalCatalogFactory
