Merge branch '1.8_release_3.9.x' into hotfix_3.9.x_jingdongfang

Author: kanata163

flinkx-hdfs/flinkx-hdfs-core/src/main/java/com/dtstack/flinkx/hdfs/HdfsUtil.java

@@ -40,7 +40,7 @@ public class HdfsUtil {
     public static final String NULL_VALUE = "\\N";
 
     public static Object string2col(String str, String type, SimpleDateFormat customDateFormat) {
-        if (str == null || str.length() == 0){
+        if (str == null){
             return null;
         }
 

flinkx-hdfs/flinkx-hdfs-reader/src/main/java/com/dtstack/flinkx/hdfs/reader/HdfsTextInputFormat.java

@@ -22,6 +22,7 @@
 import com.dtstack.flinkx.reader.MetaColumn;
 import jodd.util.StringUtil;
 import org.apache.commons.io.output.ByteArrayOutputStream;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.core.io.InputSplit;
 import org.apache.flink.types.Row;
 import org.apache.hadoop.fs.Path;
@@ -85,7 +86,7 @@ public void openInternal(InputSplit inputSplit) throws IOException {
     @Override
     public Row nextRecordInternal(Row row) throws IOException {
         String line = new String(((Text)value).getBytes(), 0, ((Text)value).getLength(), charsetName);
-        String[] fields = line.split(delimiter);
+        String[] fields = StringUtils.splitPreserveAllTokens(line, delimiter);
 
         if (metaColumns.size() == 1 && "*".equals(metaColumns.get(0).getName())){
             row = new Row(fields.length);

flinkx-mongodb/flinkx-mongodb-core/src/main/java/com/dtstack/flinkx/mongodb/MongodbConfigKeys.java

@@ -57,4 +57,6 @@ public class MongodbConfigKeys {
     public final static String KEY_MAX_WAIT_TIME = "maxWaitTime";
 
     public final static String KEY_SOCKET_TIMEOUT = "socketTimeout";
+
+    public final static String KEY_SERVER_SELECTION_TIMEOUT = "serverSelectionTimeout";
 }

flinkx-mongodb/flinkx-mongodb-core/src/main/java/com/dtstack/flinkx/mongodb/MongodbUtil.java

@@ -70,6 +70,8 @@ public class MongodbUtil {
 
     private static  final Integer DEFAULT_SOCKET_TIMEOUT = 0;
 
+    private static  final Integer DEFAULT_SERVER_SELECTION_TIMEOUT = 5 * 1000;
+
     /**
      * Get mongo client
      * @param mongodbConfig
@@ -96,7 +98,8 @@ public static MongoClient getMongoClient(Map<String,Object> mongodbConfig){
 
                 mongoClient = new MongoClient(serverAddress,credentials,options);
             }
-
+            //验证client是否连接成功
+            mongoClient.listDatabases();
             LOG.info("Get mongodb client successful");
             return mongoClient;
         }catch (Exception e){
@@ -196,6 +199,10 @@ private static MongoClientOptions getOption(Map<String,Object> mongodbConfig){
         LOG.info("Mongodb config -- socketTimeout:" + socketTimeout);
         build.maxWaitTime(socketTimeout);
 
+        int serverSelectionTimeout = MapUtils.getIntValue(mongodbConfig, KEY_SERVER_SELECTION_TIMEOUT, DEFAULT_SERVER_SELECTION_TIMEOUT);
+        LOG.info("Mongodb config -- serverSelectionTimeout:" + serverSelectionTimeout);
+        build.serverSelectionTimeout(serverSelectionTimeout);
+
         build.writeConcern(WriteConcern.UNACKNOWLEDGED);
         return build.build();
     }

flinkx-mongodb/flinkx-mongodb-reader/src/main/java/com/dtstack/flinkx/mongodb/reader/MongodbInputFormat.java

@@ -22,6 +22,7 @@
 import com.dtstack.flinkx.mongodb.MongodbConfigKeys;
 import com.dtstack.flinkx.mongodb.MongodbUtil;
 import com.dtstack.flinkx.reader.MetaColumn;
+import com.dtstack.flinkx.util.ExceptionUtil;
 import com.dtstack.flinkx.util.StringUtil;
 import com.mongodb.BasicDBObject;
 import com.mongodb.MongoClient;
@@ -38,7 +39,9 @@
 import org.bson.conversions.Bson;
 
 import java.io.IOException;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
 
 /**
  * Read plugin for reading static data
@@ -171,7 +174,8 @@ public InputSplit[] createInputSplits(int minNumSplits) throws IOException {
                 splits.add(new MongodbInputSplit((int)(size * minNumSplits), (int)(docNum - size * minNumSplits)));
             }
         } catch (Exception e){
-            LOG.error("{}", e);
+            LOG.error("error to create inputSplits, e = {}", ExceptionUtil.getErrorMessage(e));
+            throw e;
         } finally {
             MongodbUtil.close(client, null);
         }

flinkx-rdb/flinkx-rdb-reader/src/main/java/com.dtstack.flinkx.rdb.inputformat/DistributedJdbcInputFormat.java

@@ -206,7 +206,7 @@ protected void closeCurrentSource(){
 
     @Override
     protected void closeInternal() throws IOException {
-
+        closeCurrentSource();
     }
 
     @Override

flinkx-test/pom.xml

@@ -30,12 +30,6 @@
             <version>19.0</version>
         </dependency>
 
-        <dependency>
-            <groupId>ch.qos.logback</groupId>
-            <artifactId>logback-classic</artifactId>
-            <version>1.1.7</version>
-        </dependency>
-
         <dependency>
             <groupId>com.google.code.gson</groupId>
             <artifactId>gson</artifactId>