[feat-35399]1.[http] 使用PoolingHttpClientConnectionManager来创建连接线程池; 2.[file] 优化File source 代码，移除location, hdfsSite, coreSite等不必要参数，调整FileSourceTableInfo的结构，将Csv、Json、Arvo等拆分为独立类

Author: FlechazoW

core/src/main/java/com/dtstack/flink/sql/util/DtStringUtil.java

@@ -21,10 +21,11 @@
 package com.dtstack.flink.sql.util;
 
 import com.dtstack.flink.sql.enums.ColumnType;
-import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.base.Strings;
 import com.google.common.collect.Maps;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.flink.util.Preconditions;
 
 import java.math.BigDecimal;
 import java.sql.Timestamp;
@@ -46,7 +47,7 @@ public class DtStringUtil {
 
     private static final Pattern NO_VERSION_PATTERN = Pattern.compile("([a-zA-Z]+).*");
 
-    private static ObjectMapper objectMapper = new ObjectMapper();
+    private static final ObjectMapper objectMapper = new ObjectMapper();
 
     /**
      * Split the specified string delimiter --- ignored quotes delimiter
@@ -253,11 +254,11 @@ public static String col2string(Object column, String type) {
     }
 
     public static String getPluginTypeWithoutVersion(String engineType) {
+        Preconditions.checkNotNull(engineType, "type can't be null!");
+
         Matcher matcher = NO_VERSION_PATTERN.matcher(engineType);
 
-        if (!engineType.equals("kafka")) {
-            return engineType;
-        } else if (!matcher.find()) {
+        if (!matcher.find()) {
             return engineType;
         }
 

core/src/main/java/com/dtstack/flink/sql/util/PluginUtil.java

@@ -22,11 +22,9 @@
 
 import com.dtstack.flink.sql.dirtyManager.consumer.DirtyConsumerFactory;
 import com.dtstack.flink.sql.enums.EPluginLoadMode;
-import com.fasterxml.jackson.core.JsonGenerationException;
-import com.fasterxml.jackson.core.JsonParseException;
-import com.fasterxml.jackson.databind.JsonMappingException;
-import com.fasterxml.jackson.databind.ObjectMapper;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonProcessingException;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -128,7 +126,11 @@ public static Map<String,Object> objectToMap(Object obj) throws Exception{
         return objectMapper.readValue(objectMapper.writeValueAsBytes(obj), Map.class);
     }
 
-    public static <T> T jsonStrToObject(String jsonStr, Class<T> clazz) throws JsonParseException, JsonMappingException, JsonGenerationException, IOException{
+    public static String objectToString(Object obj) throws JsonProcessingException {
+        return objectMapper.writeValueAsString(obj);
+    }
+
+    public static <T> T jsonStrToObject(String jsonStr, Class<T> clazz) throws IOException{
         return  objectMapper.readValue(jsonStr, clazz);
     }
 

core/src/main/java/com/dtstack/flink/sql/util/ThreadUtil.java

@@ -27,6 +27,7 @@
  */
 public class ThreadUtil {
     public static final Long DEFAULT_SLEEP_TIME = 10L;
+
     public static void sleepSeconds(long timeout) {
         try {
             TimeUnit.SECONDS.sleep(timeout);
@@ -50,4 +51,12 @@ public static void sleepMicroseconds(long timeout) {
             throw new RuntimeException(ie);
         }
     }
+
+    public static void sleepMilliseconds(long timeout) {
+        try {
+            TimeUnit.MILLISECONDS.sleep(timeout);
+        } catch (InterruptedException e) {
+            throw new RuntimeException(e);
+        }
+    }
 }

file/file-source/src/main/java/com/dtstack/flink/sql/source/file/FileSource.java

@@ -23,7 +23,6 @@
 import com.dtstack.flink.sql.table.AbstractSourceTableInfo;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.api.common.serialization.DeserializationSchema;
-import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.runtime.execution.SuppressRestartsException;
 import org.apache.flink.streaming.api.datastream.DataStreamSource;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
@@ -40,12 +39,13 @@
 import java.io.BufferedReader;
 import java.io.File;
 import java.io.FileInputStream;
+import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.InputStreamReader;
 import java.net.URI;
 import java.util.Locale;
-import java.util.Objects;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 /**
  * @author tiezhu
@@ -63,113 +63,108 @@ public class FileSource implements IStreamSourceGener<Table>, SourceFunction<Row
     /**
      * Flag to mark the main work loop as alive.
      */
-    private volatile boolean running = true;
+    private final AtomicBoolean running = new AtomicBoolean(true);
 
-    private FileSourceTableInfo fileSourceTableInfo;
+    private URI fileUri;
 
     private InputStream inputStream;
 
-    public void setDeserializationSchema(DeserializationSchema<Row> deserializationSchema) {
-        this.deserializationSchema = deserializationSchema;
-    }
-
-    public void setFileSourceTableInfo(FileSourceTableInfo fileSourceTableInfo) {
-        this.fileSourceTableInfo = fileSourceTableInfo;
-    }
-
-    public FileSourceTableInfo getFileSourceTableInfo() {
-        return fileSourceTableInfo;
-    }
+    private String charset;
 
     @Override
     public Table genStreamSource(AbstractSourceTableInfo sourceTableInfo,
                                  StreamExecutionEnvironment env,
                                  StreamTableEnvironment tableEnv) {
         FileSource fileSource = new FileSource();
         FileSourceTableInfo tableInfo = (FileSourceTableInfo) sourceTableInfo;
-        fileSource.setFileSourceTableInfo(tableInfo);
-        fileSource.setDeserializationSchema(tableInfo.buildDeserializationSchema());
-
-        TypeInformation<Row> rowTypeInformation = tableInfo.getRowTypeInformation();
-        String operatorName = tableInfo.getOperatorName();
 
-        DataStreamSource<?> source = env.addSource(fileSource, operatorName, rowTypeInformation);
+        DataStreamSource<?> source = fileSource.initDataStream(tableInfo, env);
+        String fields = StringUtils.join(tableInfo.getFields(), ",");
 
-        String fields = StringUtils.join(fileSource.getFileSourceTableInfo().getFields(), ",");
         return tableEnv.fromDataStream(source, fields);
     }
 
+    public DataStreamSource<?> initDataStream(FileSourceTableInfo tableInfo,
+                                              StreamExecutionEnvironment env) {
+        deserializationSchema = tableInfo.getDeserializationSchema();
+        fileUri = URI.create(tableInfo.getFilePath() + SP + tableInfo.getFileName());
+        charset = tableInfo.getCharsetName();
+        return env.addSource(
+            this,
+            tableInfo.getOperatorName(),
+            tableInfo.buildRowTypeInfo());
+    }
+
     /**
      * 根据存储位置的不同，获取不同的input stream
      *
-     * @param sourceTableInfo source table info
+     * @param fileUri file uri
      * @return input stream
-     * @throws Exception reader exception
      */
-    private InputStream getInputStream(FileSourceTableInfo sourceTableInfo) throws Exception {
-        switch (sourceTableInfo.getLocation().toLowerCase(Locale.ROOT)) {
-            case "local":
-                return fromLocalFile(sourceTableInfo);
-            case "hdfs":
-                return fromHdfsFile(sourceTableInfo);
-            default:
-                throw new IllegalArgumentException();
+    private InputStream getInputStream(URI fileUri) {
+        try {
+            String scheme = fileUri.getScheme() == null ? "local" : fileUri.getScheme();
+            switch (scheme.toLowerCase(Locale.ROOT)) {
+                case "local":
+                    return fromLocalFile(fileUri);
+                case "hdfs":
+                    return fromHdfsFile(fileUri);
+                default:
+                    throw new UnsupportedOperationException(
+                        String.format("Unsupported type [%s] of file.", scheme)
+                    );
+            }
+        } catch (IOException e) {
+            throw new SuppressRestartsException(e);
         }
     }
 
     /**
      * 从HDFS上获取文件内容
      *
-     * @param sourceTableInfo source table info
+     * @param fileUri file uri of file
      * @return hdfs file input stream
-     * @throws Exception reader exception
+     * @throws IOException reader exception
      */
-    private InputStream fromHdfsFile(FileSourceTableInfo sourceTableInfo) throws Exception {
-        String filePath = sourceTableInfo.getFilePath();
-        String fileName = sourceTableInfo.getFileName();
-        String path = filePath + SP + fileName;
-
+    private InputStream fromHdfsFile(URI fileUri) throws IOException {
         Configuration conf = new Configuration();
-        conf.addResource(new Path(sourceTableInfo.getHdfsSite()));
-        conf.addResource(new Path(sourceTableInfo.getCoreSite()));
-        FileSystem fs = FileSystem.newInstance(new URI(filePath), conf, sourceTableInfo.getHdfsUser());
-        return fs.open(new Path(path));
+
+        // get conf from HADOOP_CONF_DIR
+        String hadoopConfDir = System.getenv("HADOOP_CONF_DIR");
+        String confHome = hadoopConfDir == null ? "." : hadoopConfDir;
+
+        conf.addResource(new Path(confHome + SP + "hdfs-site.xml"));
+
+        FileSystem fs = FileSystem.get(fileUri, conf);
+        return fs.open(new Path(fileUri.getPath()));
     }
 
     /**
      * 从本地获取文件内容
      *
-     * @param sourceTableInfo source table
+     * @param fileUri file uri of file
      * @return local file input stream
-     * @throws Exception read exception
+     * @throws FileNotFoundException read exception
      */
-    private InputStream fromLocalFile(FileSourceTableInfo sourceTableInfo) throws Exception {
-        String filePath = sourceTableInfo.getFilePath();
-        String fileName = sourceTableInfo.getFileName();
-        String path = filePath + SP + fileName;
-        File file = new File(path);
+    private InputStream fromLocalFile(URI fileUri) throws FileNotFoundException {
+        File file = new File(fileUri.getPath());
         if (file.exists()) {
             return new FileInputStream(file);
         } else {
             throw new SuppressRestartsException(new IOException(
-                String.format(
-                    "File [%s] not exist. File path: [%s]",
-                    sourceTableInfo.getFileName(),
-                    sourceTableInfo.getFilePath())
-            ));
+                String.format("File not exist. File path: [%s]", fileUri.getPath())));
         }
     }
 
-
     @Override
     public void run(SourceContext<Row> ctx) throws Exception {
-        inputStream = getInputStream(fileSourceTableInfo);
-        BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(inputStream));
+        inputStream = getInputStream(fileUri);
+        BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(inputStream, charset));
 
-        while (running) {
+        while (running.get()) {
             String line = bufferedReader.readLine();
             if (line == null) {
-                running = false;
+                running.compareAndSet(true, false);
                 inputStream.close();
                 break;
             } else {
@@ -182,8 +177,8 @@ public void run(SourceContext<Row> ctx) throws Exception {
     @Override
     public void cancel() {
         LOG.info("File source cancel..");
-        running = false;
-        if (Objects.nonNull(inputStream)) {
+        running.compareAndSet(true, false);
+        if (inputStream != null) {
             try {
                 inputStream.close();
             } catch (IOException ioException) {

file/file-source/src/main/java/com/dtstack/flink/sql/source/file/FileSourceConstant.java

@@ -0,0 +1,69 @@
+/*
+ * 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.source.file;
+
+import java.io.File;
+
+/**
+ * @author tiezhu
+ * @date 2021/3/22 星期一
+ * Company dtstack
+ */
+public class FileSourceConstant {
+    public static final String TYPE_KEY = "type";
+
+    public static final String FORMAT_KEY = "format";
+
+    public static final String FIELD_DELIMITER_KEY = "fieldDelimiter";
+
+    public static final String FILE_NAME_KEY = "fileName";
+
+    public static final String FILE_PATH_KEY = "filePath";
+
+    public static final String IGNORE_PARSER_ERROR = "ignoreParserError";
+
+    public static final String CHARSET_NAME_KEY = "charsetName";
+
+    public static final String AVRO_FORMAT_KEY = "avroFormat";
+
+    public static final String NULL_LITERAL_KEY = "nullLiteral";
+
+    public static final String ALLOW_COMMENT_KEY = "allowComment";
+
+    public static final String ARRAY_ELEMENT_DELIMITER_KEY = "arrayElementDelimiter";
+
+    public static final String QUOTA_CHARACTER_KEY = "quoteCharacter";
+
+    public static final String ESCAPE_CHARACTER_KEY = "escapeCharacter";
+
+    public static final Character DEFAULT_DELIMITER = ',';
+
+    public static final String DEFAULT_FILE_FORMAT = "csv";
+
+    public static final String DEFAULT_CHARSET = "UTF-8";
+
+    public static final String DEFAULT_PATH = ".";
+
+    public static final char DEFAULT_QUOTE_CHAR = '"';
+
+    public static final char DEFAULT_ESCAPE_CHAR = '\\';
+
+    public static final String DEFAULT_NULL_LITERAL = "null";
+
+}