[flink-table-store] branch master updated: [FLINK-30453] Fix 'can't find CatalogFactory' error when using FLINK sql-client to add table store bundle jar

[czweng]

This is an automated email from the ASF dual-hosted git repository.

czweng pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-table-store.git


The following commit(s) were added to refs/heads/master by this push:
     new 7e0d55ff [FLINK-30453] Fix 'can't find CatalogFactory' error when 
using FLINK sql-client to add table store bundle jar
7e0d55ff is described below

commit 7e0d55ff3dc9fd48455b17d9a439647b0554d020
Author: yuzelin <33053040+yuze...@users.noreply.github.com>
AuthorDate: Tue Dec 20 19:40:13 2022 +0800

    [FLINK-30453] Fix 'can't find CatalogFactory' error when using FLINK 
sql-client to add table store bundle jar
    
    This closes #442.
---
 .../flink/table/store/connector/FlinkCatalogFactory.java     | 12 +++++++++---
 .../apache/flink/table/store/connector/FlinkCatalogTest.java |  4 +++-
 .../flink/table/store/file/catalog/CatalogFactory.java       | 10 +++++++++-
 3 files changed, 21 insertions(+), 5 deletions(-)

diff --git 
a/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/FlinkCatalogFactory.java
 
b/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/FlinkCatalogFactory.java
index 9c70ffe9..43665476 100644
--- 
a/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/FlinkCatalogFactory.java
+++ 
b/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/FlinkCatalogFactory.java
@@ -54,11 +54,17 @@ public class FlinkCatalogFactory implements 
org.apache.flink.table.factories.Cat
 
     @Override
     public FlinkCatalog createCatalog(Context context) {
-        return createCatalog(context.getName(), 
Configuration.fromMap(context.getOptions()));
+        return createCatalog(
+                context.getName(),
+                Configuration.fromMap(context.getOptions()),
+                context.getClassLoader());
     }
 
-    public static FlinkCatalog createCatalog(String catalogName, Configuration 
options) {
+    public static FlinkCatalog createCatalog(
+            String catalogName, Configuration options, ClassLoader 
classLoader) {
         return new FlinkCatalog(
-                CatalogFactory.createCatalog(options), catalogName, 
options.get(DEFAULT_DATABASE));
+                CatalogFactory.createCatalog(options, classLoader),
+                catalogName,
+                options.get(DEFAULT_DATABASE));
     }
 }
diff --git 
a/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/FlinkCatalogTest.java
 
b/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/FlinkCatalogTest.java
index e7a6d6fe..93271875 100644
--- 
a/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/FlinkCatalogTest.java
+++ 
b/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/FlinkCatalogTest.java
@@ -75,7 +75,9 @@ public class FlinkCatalogTest {
         String path = TEMPORARY_FOLDER.newFolder().toURI().toString();
         Configuration conf = new Configuration();
         conf.setString("warehouse", path);
-        catalog = FlinkCatalogFactory.createCatalog("test-catalog", conf);
+        catalog =
+                FlinkCatalogFactory.createCatalog(
+                        "test-catalog", conf, 
FlinkCatalogTest.class.getClassLoader());
     }
 
     private ResolvedSchema createSchema() {
diff --git 
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/catalog/CatalogFactory.java
 
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/catalog/CatalogFactory.java
index 66129710..53a68e02 100644
--- 
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/catalog/CatalogFactory.java
+++ 
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/catalog/CatalogFactory.java
@@ -49,7 +49,15 @@ public interface CatalogFactory {
         return new Path(warehouse);
     }
 
+    /**
+     * If the ClassLoader is not specified, using the context ClassLoader of 
current thread as
+     * default.
+     */
     static Catalog createCatalog(Configuration options) {
+        return createCatalog(options, 
Thread.currentThread().getContextClassLoader());
+    }
+
+    static Catalog createCatalog(Configuration options, ClassLoader 
classLoader) {
         // manual validation
         // because different catalog types may have different options
         // we can't list them all in the optionalOptions() method
@@ -57,7 +65,7 @@ public interface CatalogFactory {
 
         String metastore = options.get(METASTORE);
         List<CatalogFactory> factories = new ArrayList<>();
-        ServiceLoader.load(CatalogFactory.class, 
Thread.currentThread().getContextClassLoader())
+        ServiceLoader.load(CatalogFactory.class, classLoader)
                 .iterator()
                 .forEachRemaining(
                         f -> {