add log

Author: yanxi0227

core/src/main/java/com/dtstack/flink/sql/Main.java

@@ -173,7 +173,9 @@ public static void main(String[] args) throws Exception {
         }
 
         for (InsertSqlParser.SqlParseResult result : sqlTree.getExecSqlList()) {
-            LOG.info("exe-sql:\n" + result.getExecSql());
+            if(LOG.isInfoEnabled()){
+                LOG.info("exe-sql:\n" + result.getExecSql());
+            }
 
             boolean isSide = false;
 
@@ -203,7 +205,9 @@ public static void main(String[] args) throws Exception {
                         sideSqlExec.exec(result.getExecSql(), sideTableMap, tableEnv, registerTableCache);
                     }else{
                         tableEnv.sqlUpdate(result.getExecSql());
-                        LOG.info("exec sql: " + result.getExecSql());
+                        if(LOG.isInfoEnabled()){
+
+                        }LOG.info("exec sql: " + result.getExecSql());
                     }
                 }
             }
@@ -286,7 +290,9 @@ private static void registerTable(SqlTree sqlTree, StreamExecutionEnvironment en
 
                 Table regTable = tableEnv.fromDataStream(adaptStream, fields);
                 tableEnv.registerTable(tableInfo.getName(), regTable);
-                LOG.info("registe table {} success.", tableInfo.getName());
+                if(LOG.isInfoEnabled()){
+                    LOG.info("registe table {} success.", tableInfo.getName());
+                }
                 registerTableCache.put(tableInfo.getName(), regTable);
                 classPathSet.add(PluginUtil.getRemoteJarFilePath(tableInfo.getType(), SourceTableInfo.SOURCE_SUFFIX, remoteSqlPluginPath, localSqlPluginPath));
             } else if (tableInfo instanceof TargetTableInfo) {

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

@@ -106,7 +106,9 @@ public void exec(String sql, Map<String, SideTableInfo> sideTableMap, StreamTabl
 
                 if(pollSqlNode.getKind() == INSERT){
                     tableEnv.sqlUpdate(pollSqlNode.toString());
-                    LOG.info("exec sql: " + pollSqlNode.toString());
+                    if(LOG.isInfoEnabled()){
+                        LOG.info("exec sql: " + pollSqlNode.toString());
+                    }
                 }else if(pollSqlNode.getKind() == AS){
                     AliasInfo aliasInfo = parseASNode(pollSqlNode);
                     Table table = tableEnv.sql(aliasInfo.getName());
@@ -536,7 +538,9 @@ public void registerTmpTable(CreateTmpTableParser.SqlParserResult result,
                     AliasInfo aliasInfo = parseASNode(pollSqlNode);
                     Table table = tableEnv.sql(aliasInfo.getName());
                     tableEnv.registerTable(aliasInfo.getAlias(), table);
-                    LOG.info("Register Table {} by {}", aliasInfo.getAlias(), aliasInfo.getName());
+                    if(LOG.isInfoEnabled()){
+                        LOG.info("Register Table {} by {}", aliasInfo.getAlias(), aliasInfo.getName());
+                    }
                     localTableCache.put(aliasInfo.getAlias(), table);
                 } else if (pollSqlNode.getKind() == SELECT){
                     Table table = tableEnv.sqlQuery(pollObj.toString());

kafka09/kafka09-source/src/main/java/com/dtstack/flink/sql/source/kafka/table/KafkaSourceParser.java

@@ -63,7 +63,9 @@ static void dealNestField(Matcher matcher, TableInfo tableInfo) {
         tableInfo.addField(mappingField);
         tableInfo.addFieldClass(fieldClass);
         tableInfo.addFieldType(fieldType);
-        LOG.info(physicalField + "--->" + mappingField + " Class: " + fieldClass.toString());
+        if(LOG.isInfoEnabled()){
+            LOG.info(physicalField + "--->" + mappingField + " Class: " + fieldClass.toString());
+        }
     }
 
     @Override

kafka10/kafka10-source/src/main/java/com/dtstack/flink/sql/source/kafka/table/KafkaSourceParser.java

@@ -24,6 +24,8 @@
 import com.dtstack.flink.sql.table.TableInfo;
 import com.dtstack.flink.sql.util.ClassUtil;
 import com.dtstack.flink.sql.util.MathUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.Map;
 import java.util.regex.Matcher;
@@ -38,6 +40,8 @@
 
 public class KafkaSourceParser extends AbsSourceParser {
 
+    private static final Logger LOG = LoggerFactory.getLogger(KafkaSourceParser.class);
+
     private static final String KAFKA_NEST_FIELD_KEY = "nestFieldKey";
 
     private static Pattern kafkaNestFieldKeyPattern = Pattern.compile("(?i)((@*\\w+\\.)*\\w+)\\s+(\\w+)\\s+AS\\s+(\\w+)$");
@@ -63,6 +67,9 @@ static void dealNestField(Matcher matcher, TableInfo tableInfo) {
         tableInfo.addField(mappingField);
         tableInfo.addFieldClass(fieldClass);
         tableInfo.addFieldType(fieldType);
+        if(LOG.isInfoEnabled()){
+            LOG.info(physicalField + "--->" + mappingField + " Class: " + fieldClass.toString());
+        }
     }
 
     @Override

kafka11/kafka11-source/src/main/java/com/dtstack/flink/sql/source/kafka/table/KafkaSourceParser.java

@@ -24,6 +24,8 @@
 import com.dtstack.flink.sql.table.TableInfo;
 import com.dtstack.flink.sql.util.ClassUtil;
 import com.dtstack.flink.sql.util.MathUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.Map;
 import java.util.regex.Matcher;
@@ -38,6 +40,8 @@
 
 public class KafkaSourceParser extends AbsSourceParser {
 
+    private static final Logger LOG = LoggerFactory.getLogger(KafkaSourceParser.class);
+
     private static final String KAFKA_NEST_FIELD_KEY = "nestFieldKey";
 
     private static Pattern kafkaNestFieldKeyPattern = Pattern.compile("(?i)((@*\\w+\\.)*\\w+)\\s+(\\w+)\\s+AS\\s+(\\w+)$");
@@ -63,6 +67,9 @@ static void dealNestField(Matcher matcher, TableInfo tableInfo) {
         tableInfo.addField(mappingField);
         tableInfo.addFieldClass(fieldClass);
         tableInfo.addFieldType(fieldType);
+        if(LOG.isInfoEnabled()){
+            LOG.info(physicalField + "--->" + mappingField + " Class: " + fieldClass.toString());
+        }
     }
 
     @Override