codeview

Author: dapeng

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

@@ -98,9 +98,9 @@ public static List<URL> getClassPath() {
 
 
 
-    public static URLClassLoader loadExtraJar(List<URL> jarURLList, URLClassLoader classLoader)
+    public static URLClassLoader loadExtraJar(List<URL> jarUrlList, URLClassLoader classLoader)
             throws  IllegalAccessException, InvocationTargetException {
-        for(URL url : jarURLList){
+        for(URL url : jarUrlList){
             if(url.toString().endsWith(".jar")){
                 urlClassLoaderAddUrl(classLoader, url);
             }

core/src/main/java/com/dtstack/flink/sql/exec/ExecuteProcessHelper.java

@@ -117,7 +117,7 @@ public static ParamsInfo parseParams(String[] args) throws Exception {
         String confProp = URLDecoder.decode(options.getConfProp(), Charsets.UTF_8.toString());
         Properties confProperties = PluginUtil.jsonStrToObject(confProp, Properties.class);
 
-        List<URL> jarURList = getExternalJarUrls(options.getAddjar());
+        List<URL> jarUrlList = getExternalJarUrls(options.getAddjar());
 
         return ParamsInfo.builder()
                 .setSql(sql)
@@ -127,7 +127,7 @@ public static ParamsInfo parseParams(String[] args) throws Exception {
                 .setPluginLoadMode(pluginLoadMode)
                 .setDeployMode(deployMode)
                 .setConfProp(confProperties)
-                .setJarUrlList(jarURList)
+                .setJarUrlList(jarUrlList)
                 .setLogLevel(logLevel)
                 .build();
 

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

@@ -26,8 +26,8 @@
 import org.apache.flink.table.runtime.types.CRow;
 
 import java.sql.SQLException;
-import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 
 /**
@@ -60,7 +60,7 @@ public void open(Configuration parameters) throws Exception {
 
         //start reload cache thread
         AbstractSideTableInfo sideTableInfo = sideInfo.getSideTableInfo();
-        es = Executors.newSingleThreadScheduledExecutor(new DTThreadFactory("cache-all-reload"));
+        es = new ScheduledThreadPoolExecutor(1,new DTThreadFactory("cache-all-reload"));
         es.scheduleAtFixedRate(() -> reloadCache(), sideTableInfo.getCacheTimeout(), sideTableInfo.getCacheTimeout(), TimeUnit.MILLISECONDS);
     }
 

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

@@ -80,6 +80,7 @@ public static List<FieldInfo> getRowTypeInfo(SqlNode sqlNode, JoinScope scope, b
                 switch(identifierSize) {
                     case 1:
                         fieldInfoList.addAll(getAllField(scope));
+                        break;
                     default:
                         SqlIdentifier tableIdentify = identifier.skipLast(1);
                         JoinScope.ScopeChild scopeChild = scope.getScope(tableIdentify.getSimple());
@@ -99,6 +100,7 @@ public static List<FieldInfo> getRowTypeInfo(SqlNode sqlNode, JoinScope scope, b
                             fieldInfo.setTypeInformation(type);
                             fieldInfoList.add(fieldInfo);
                         }
+                        break;
                 }
             }
         }