fix mongo url bug

Author: dapeng

docs/mongoSide.md

@@ -30,7 +30,7 @@
  |----|---|
  | tableName | 注册到flink的表名称(可选填;不填默认和hbase对应的表名称相同)|
  | colName | 列名称|
- | colType | 列类型 [colType支持的类型](colType.md)|
+ | colType | 列类型 [colType支持的类型](docs/colType.md)|
  | PERIOD FOR SYSTEM_TIME | 关键字表明该定义的表为维表信息|
  | PRIMARY KEY(keyInfo) | 维表主键定义;多个列之间用逗号隔开|
 
@@ -62,8 +62,7 @@ create table sideTable(
     PERIOD FOR SYSTEM_TIME
  )WITH(
     type ='mongo',
-    //mongodb://[username:password@]host1[:port1][,host2[:port2],...[,hostN[:portN]]]/[?options]]
-    address ='mongodb://172.21.32.1:27017,172.21.32.1:27017',
+    address ='172.21.32.1:27017,172.21.32.1:27017',
     database ='test',
     tableName ='sidetest',
     cache ='LRU',

mongo/mongo-side/mongo-all-side/src/main/java/com/dtstack/flink/sql/side/mongo/MongoAllReqRow.java

@@ -177,9 +177,10 @@ private String buildKey(Map<String, Object> val, List<String> equalFieldList) {
         return sb.toString();
     }
 
-    private MongoCollection getConn(String address, String database, String tableName) {
+    private MongoCollection getConn(String host, String userName, String password, String database, String tableName) {
+
         MongoCollection dbCollection;
-        mongoClient = new MongoClient(new MongoClientURI(address));
+        mongoClient = new MongoClient(new MongoClientURI(getConnectionUrl(host, userName, password)));
         db = mongoClient.getDatabase(database);
         dbCollection = db.getCollection(tableName, Document.class);
         return dbCollection;
@@ -193,7 +194,8 @@ private void loadData(Map<String, List<Map<String, Object>>> tmpCache) throws SQ
         try {
             for (int i = 0; i < CONN_RETRY_NUM; i++) {
                 try {
-                    dbCollection = getConn(tableInfo.getAddress(), tableInfo.getDatabase(), tableInfo.getTableName());
+                    dbCollection = getConn(tableInfo.getAddress(), tableInfo.getUserName(), tableInfo.getPassword(),
+                            tableInfo.getDatabase(), tableInfo.getTableName());
                     break;
                 } catch (Exception e) {
                     if (i == CONN_RETRY_NUM - 1) {
@@ -251,4 +253,14 @@ private void loadData(Map<String, List<Map<String, Object>>> tmpCache) throws SQ
             }
         }
     }
+    private String getConnectionUrl(String address, String userName, String password){
+        if(address.startsWith("mongodb://") || address.startsWith("mongodb+srv://")){
+            return  address;
+        }
+        if (StringUtils.isNotBlank(userName) && StringUtils.isNotBlank(password)) {
+            return String.format("mongodb://%s:%s@%s", userName, password, address);
+        }
+        return String.format("mongodb://%s", address);
+    }
+
 }

mongo/mongo-side/mongo-async-side/src/main/java/com/dtstack/flink/sql/side/mongo/MongoAsyncReqRow.java

@@ -19,6 +19,7 @@
 
 package com.dtstack.flink.sql.side.mongo;
 
+import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.streaming.api.functions.async.ResultFuture;
@@ -84,9 +85,8 @@ public void open(Configuration parameters) throws Exception {
     }
 
     public void connMongoDb() throws Exception {
-        String address = mongoSideTableInfo.getAddress();
-        ConnectionString connectionString = new ConnectionString(address);
-
+        ConnectionString connectionString = new ConnectionString(getConnectionUrl(mongoSideTableInfo.getAddress(),
+                mongoSideTableInfo.getUserName(), mongoSideTableInfo.getPassword()));
         MongoClientSettings settings = MongoClientSettings.builder()
                 .applyConnectionString(connectionString)
                 .build();
@@ -193,4 +193,14 @@ public void close() throws Exception {
         }
     }
 
+    private String getConnectionUrl(String address, String userName, String password){
+        if(address.startsWith("mongodb://") || address.startsWith("mongodb+srv://")){
+            return  address;
+        }
+        if (StringUtils.isNotBlank(userName) && StringUtils.isNotBlank(password)) {
+            return String.format("mongodb://%s:%s@%s", userName, password, address);
+        }
+        return String.format("mongodb://%s", address);
+    }
+
 }

mongo/mongo-sink/src/main/java/com/dtstack/flink/sql/sink/mongo/MongoOutputFormat.java

@@ -25,6 +25,7 @@
 import com.mongodb.client.MongoCollection;
 import com.mongodb.client.MongoDatabase;
 import com.mongodb.client.result.UpdateResult;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.configuration.Configuration;
@@ -119,7 +120,8 @@ public void close() {
     }
 
     private void establishConnection() {
-        mongoClient = new MongoClient(new MongoClientURI(address));
+
+        mongoClient = new MongoClient(new MongoClientURI(getConnectionUrl()));
         db = mongoClient.getDatabase(database);
     }
 
@@ -197,5 +199,14 @@ public MongoOutputFormat finish() {
         }
     }
 
+    private String getConnectionUrl(){
+        if(address.startsWith("mongodb://") || address.startsWith("mongodb+srv://")){
+            return  address;
+        }
+        if (StringUtils.isNotBlank(userName) && StringUtils.isNotBlank(password)) {
+            return String.format("mongodb://%s:%s@%s", userName, password, address);
+        }
+        return String.format("mongodb://%s", address);
+    }
 
 }