flinksql 150 classloader

Author: jiemotongxue

core/src/main/java/com/dtstack/flink/sql/classloader/ClassLoaderCallBackMethod.java

@@ -0,0 +1,40 @@
+/*
+ * 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 com.dtstack.flink.sql.classloader;
+
+/**
+ * company: www.dtstack.com
+ * author: toutian
+ * create: 2019/10/14
+ */
+public class ClassLoaderCallBackMethod {
+
+    public static <R> R callbackAndReset(DtSupplier<R> supplier, ClassLoader toSetClassLoader) throws Exception {
+        ClassLoader oldClassLoader = Thread.currentThread().getContextClassLoader();
+        Thread.currentThread().setContextClassLoader(toSetClassLoader);
+        try {
+            return supplier.get(toSetClassLoader);
+        } finally {
+            Thread.currentThread().setContextClassLoader(oldClassLoader);
+        }
+    }
+
+
+}

core/src/main/java/com/dtstack/flink/sql/classloader/ClassLoaderManager.java

@@ -0,0 +1,59 @@
+/*
+ * 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 com.dtstack.flink.sql.classloader;
+
+import com.dtstack.flink.sql.util.PluginUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.net.URL;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * company: www.dtstack.com
+ * author: toutian
+ * create: 2019/10/14
+ */
+public class ClassLoaderManager {
+
+    private static final Logger LOG = LoggerFactory.getLogger(ClassLoaderManager.class);
+
+    private static Map<String, DtClassLoader> pluginClassLoader = new ConcurrentHashMap<>();
+
+    public static <R> R newInstance(String pluginJarPath, DtSupplier<R> supplier) throws Exception {
+        ClassLoader classLoader = retrieveClassLoad(pluginJarPath);
+        return ClassLoaderCallBackMethod.callbackAndReset(supplier, classLoader);
+    }
+
+    private static DtClassLoader retrieveClassLoad(String pluginJarPath) {
+        return pluginClassLoader.computeIfAbsent(pluginJarPath, k -> {
+            try {
+                URL[] urls = PluginUtil.getPluginJarUrls(pluginJarPath);
+                ClassLoader parentClassLoader = Thread.currentThread().getContextClassLoader();
+                DtClassLoader classLoader = new DtClassLoader(urls, parentClassLoader);
+                LOG.info("pluginJarPath:{} create ClassLoad successful...", pluginJarPath);
+                return classLoader;
+            } catch (Throwable e) {
+                LOG.error("retrieve ClassLoad happens error:{}", e);
+                throw new RuntimeException("retrieve ClassLoad happens error");
+            }
+        });
+    }
+}

core/src/main/java/com/dtstack/flink/sql/classloader/DtSupplier.java

@@ -0,0 +1,44 @@
+/*
+ * 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 com.dtstack.flink.sql.classloader;
+
+/**
+ * Represents a supplier of results.
+ *
+ * <p>There is no requirement that a new or distinct result be returned each
+ * time the supplier is invoked.
+ *
+ * <p>This is a <a href="package-summary.html">functional interface</a>
+ * whose functional method is {@link #get()}.
+ *
+ * @param <T> the type of results supplied by this supplier
+ *
+ * @since 1.8
+ */
+@FunctionalInterface
+public interface DtSupplier<T> {
+
+    /**
+     * Gets a result.
+     *
+     * @return a result
+     */
+    T get(ClassLoader cl) throws Exception;
+}

core/src/main/java/com/dtstack/flink/sql/side/StreamSideFactory.java

@@ -16,11 +16,10 @@
  * limitations under the License.
  */
 
- 
 
 package com.dtstack.flink.sql.side;
 
-import com.dtstack.flink.sql.classloader.DtClassLoader;
+import com.dtstack.flink.sql.classloader.ClassLoaderManager;
 import com.dtstack.flink.sql.enums.ECacheType;
 import com.dtstack.flink.sql.table.AbsSideTableParser;
 import com.dtstack.flink.sql.table.AbsTableParser;
@@ -30,6 +29,7 @@
  * get specify side parser
  * Date: 2018/7/25
  * Company: www.dtstack.com
+ *
  * @author xuchao
  */
 
@@ -40,18 +40,15 @@ public class StreamSideFactory {
     public static AbsTableParser getSqlParser(String pluginType, String sqlRootDir, String cacheType) throws Exception {
 
         String sideOperator = ECacheType.ALL.name().equals(cacheType) ? "all" : "async";
-        ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
         String pluginJarPath = PluginUtil.getSideJarFileDirPath(pluginType, sideOperator, "side", sqlRootDir);
-
-        DtClassLoader dtClassLoader = (DtClassLoader) classLoader;
-        PluginUtil.addPluginJar(pluginJarPath, dtClassLoader);
         String className = PluginUtil.getSqlParserClassName(pluginType, CURR_TYPE);
 
-        Class<?> sideParser = dtClassLoader.loadClass(className);
-        if(!AbsSideTableParser.class.isAssignableFrom(sideParser)){
-            throw new RuntimeException("class " + sideParser.getName() + " not subClass of AbsSideTableParser");
-        }
-
-        return sideParser.asSubclass(AbsTableParser.class).newInstance();
+        return ClassLoaderManager.newInstance(pluginJarPath, (cl) -> {
+            Class<?> sideParser = cl.loadClass(className);
+            if (!AbsSideTableParser.class.isAssignableFrom(sideParser)) {
+                throw new RuntimeException("class " + sideParser.getName() + " not subClass of AbsSideTableParser");
+            }
+            return sideParser.asSubclass(AbsTableParser.class).newInstance();
+        });
     }
 }

core/src/main/java/com/dtstack/flink/sql/side/operator/SideAsyncOperator.java

@@ -19,7 +19,7 @@
 
 package com.dtstack.flink.sql.side.operator;
 
-import com.dtstack.flink.sql.classloader.DtClassLoader;
+import com.dtstack.flink.sql.classloader.ClassLoaderManager;
 import com.dtstack.flink.sql.side.AsyncReqRow;
 import com.dtstack.flink.sql.side.FieldInfo;
 import com.dtstack.flink.sql.side.JoinInfo;
@@ -49,14 +49,13 @@ public class SideAsyncOperator {
 
     private static AsyncReqRow loadAsyncReq(String sideType, String sqlRootDir, RowTypeInfo rowTypeInfo,
                                             JoinInfo joinInfo, List<FieldInfo> outFieldInfoList, SideTableInfo sideTableInfo) throws Exception {
-        ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
         String pathOfType = String.format(PATH_FORMAT, sideType);
         String pluginJarPath = PluginUtil.getJarFileDirPath(pathOfType, sqlRootDir);
-        DtClassLoader dtClassLoader = (DtClassLoader) classLoader;
-        PluginUtil.addPluginJar(pluginJarPath, dtClassLoader);
         String className = PluginUtil.getSqlSideClassName(sideType, "side", "Async");
-        return dtClassLoader.loadClass(className).asSubclass(AsyncReqRow.class)
-                .getConstructor(RowTypeInfo.class, JoinInfo.class, List.class, SideTableInfo.class).newInstance(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo);
+        return ClassLoaderManager.newInstance(pluginJarPath, (cl) ->
+                cl.loadClass(className).asSubclass(AsyncReqRow.class)
+                        .getConstructor(RowTypeInfo.class, JoinInfo.class, List.class, SideTableInfo.class)
+                        .newInstance(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo));
     }
 
     public static DataStream getSideJoinDataStream(DataStream inputStream, String sideType, String sqlRootDir, RowTypeInfo rowTypeInfo,  JoinInfo joinInfo,

core/src/main/java/com/dtstack/flink/sql/side/operator/SideWithAllCacheOperator.java

@@ -19,7 +19,7 @@
 
 package com.dtstack.flink.sql.side.operator;
 
-import com.dtstack.flink.sql.classloader.DtClassLoader;
+import com.dtstack.flink.sql.classloader.ClassLoaderManager;
 import com.dtstack.flink.sql.side.AllReqRow;
 import com.dtstack.flink.sql.side.FieldInfo;
 import com.dtstack.flink.sql.side.JoinInfo;
@@ -28,7 +28,6 @@
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
 import org.apache.flink.streaming.api.datastream.DataStream;
 
-import java.net.MalformedURLException;
 import java.util.List;
 
 /**
@@ -47,18 +46,13 @@ private static AllReqRow loadFlatMap(String sideType, String sqlRootDir, RowType
                                          JoinInfo joinInfo, List<FieldInfo> outFieldInfoList,
                                          SideTableInfo sideTableInfo) throws Exception {
 
-        ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
         String pathOfType = String.format(PATH_FORMAT, sideType);
         String pluginJarPath = PluginUtil.getJarFileDirPath(pathOfType, sqlRootDir);
-
-        DtClassLoader dtClassLoader = (DtClassLoader) classLoader;
-        PluginUtil.addPluginJar(pluginJarPath, dtClassLoader);
         String className = PluginUtil.getSqlSideClassName(sideType, "side", "All");
 
-        return dtClassLoader.loadClass(className).asSubclass(AllReqRow.class).getConstructor(RowTypeInfo.class, JoinInfo.class, List.class, SideTableInfo.class)
-                .newInstance(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo);
-
-
+        return ClassLoaderManager.newInstance(pluginJarPath, (cl) -> cl.loadClass(className).asSubclass(AllReqRow.class)
+                .getConstructor(RowTypeInfo.class, JoinInfo.class, List.class, SideTableInfo.class)
+                .newInstance(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo));
     }
 
     public static DataStream getSideJoinDataStream(DataStream inputStream, String sideType, String sqlRootDir, RowTypeInfo rowTypeInfo, JoinInfo joinInfo,

core/src/main/java/com/dtstack/flink/sql/sink/StreamSinkFactory.java

@@ -20,12 +20,11 @@
 
 package com.dtstack.flink.sql.sink;
 
-import com.dtstack.flink.sql.classloader.DtClassLoader;
+import com.dtstack.flink.sql.classloader.ClassLoaderManager;
 import com.dtstack.flink.sql.table.AbsTableParser;
 import com.dtstack.flink.sql.table.TargetTableInfo;
 import com.dtstack.flink.sql.util.DtStringUtil;
 import com.dtstack.flink.sql.util.PluginUtil;
-import org.apache.flink.calcite.shaded.com.google.common.collect.Lists;
 import org.apache.flink.table.sinks.TableSink;
 
 /**
@@ -42,51 +41,33 @@ public class StreamSinkFactory {
     private static final String DIR_NAME_FORMAT = "%ssink";
 
     public static AbsTableParser getSqlParser(String pluginType, String sqlRootDir) throws Exception {
-        ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
-
-        if(!(classLoader instanceof DtClassLoader)){
-            throw new RuntimeException("it's not a correct classLoader instance, it's type must be DtClassLoader!");
-        }
-
-        DtClassLoader dtClassLoader = (DtClassLoader) classLoader;
-
         String pluginJarPath = PluginUtil.getJarFileDirPath(String.format(DIR_NAME_FORMAT, pluginType), sqlRootDir);
-        PluginUtil.addPluginJar(pluginJarPath, dtClassLoader);
         String typeNoVersion = DtStringUtil.getPluginTypeWithoutVersion(pluginType);
         String className = PluginUtil.getSqlParserClassName(typeNoVersion, CURR_TYPE);
-        Class<?> targetParser = dtClassLoader.loadClass(className);
 
-        if(!AbsTableParser.class.isAssignableFrom(targetParser)){
-            throw new RuntimeException("class " + targetParser.getName() + " not subClass of AbsTableParser");
-        }
-
-        return targetParser.asSubclass(AbsTableParser.class).newInstance();
+        return ClassLoaderManager.newInstance(pluginJarPath, (cl) -> {
+            Class<?> targetParser = cl.loadClass(className);
+            if(!AbsTableParser.class.isAssignableFrom(targetParser)){
+                throw new RuntimeException("class " + targetParser.getName() + " not subClass of AbsTableParser");
+            }
+            return targetParser.asSubclass(AbsTableParser.class).newInstance();
+        });
     }
 
     public static TableSink getTableSink(TargetTableInfo targetTableInfo, String localSqlRootDir) throws Exception {
-
-        ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
-        if(!(classLoader instanceof DtClassLoader)){
-            throw new RuntimeException("it's not a correct classLoader instance, it's type must be DtClassLoader!");
-        }
-
-        DtClassLoader dtClassLoader = (DtClassLoader) classLoader;
-
         String pluginType = targetTableInfo.getType();
         String pluginJarDirPath = PluginUtil.getJarFileDirPath(String.format(DIR_NAME_FORMAT, pluginType), localSqlRootDir);
-
-        PluginUtil.addPluginJar(pluginJarDirPath, dtClassLoader);
-
         String typeNoVersion = DtStringUtil.getPluginTypeWithoutVersion(pluginType);
         String className = PluginUtil.getGenerClassName(typeNoVersion, CURR_TYPE);
-        Class<?> sinkClass = dtClassLoader.loadClass(className);
-
-        if(!IStreamSinkGener.class.isAssignableFrom(sinkClass)){
-            throw new RuntimeException("class " + sinkClass + " not subClass of IStreamSinkGener");
-        }
 
-        IStreamSinkGener streamSinkGener = sinkClass.asSubclass(IStreamSinkGener.class).newInstance();
-        Object result = streamSinkGener.genStreamSink(targetTableInfo);
-        return (TableSink) result;
+        return ClassLoaderManager.newInstance(pluginJarDirPath, (cl) -> {
+            Class<?> sinkClass = cl.loadClass(className);
+            if(!IStreamSinkGener.class.isAssignableFrom(sinkClass)){
+                throw new RuntimeException("class " + sinkClass + " not subClass of IStreamSinkGener");
+            }
+            IStreamSinkGener streamSinkGener = sinkClass.asSubclass(IStreamSinkGener.class).newInstance();
+            Object result = streamSinkGener.genStreamSink(targetTableInfo);
+            return (TableSink) result;
+        });
     }
 }