rdb async connection pool size

todd5167 · todd5167 · commit 4a4f21c0d31f · 2020-04-10T12:00:19.000+08:00
diff --git a/clickhouse/clickhouse-side/clickhouse-async-side/src/main/java/com/dtstack/flink/sql/side/clickhouse/ClickhouseAsyncReqRow.java b/clickhouse/clickhouse-side/clickhouse-async-side/src/main/java/com/dtstack/flink/sql/side/clickhouse/ClickhouseAsyncReqRow.java
@@ -48,7 +48,7 @@ public void open(Configuration parameters) throws Exception {
         RdbSideTableInfo rdbSideTableInfo = (RdbSideTableInfo) sideInfo.getSideTableInfo();
         clickhouseClientConfig.put("url", rdbSideTableInfo.getUrl())
                 .put("driver_class", CLICKHOUSE_DRIVER)
-                .put("max_pool_size", DEFAULT_MAX_DB_CONN_POOL_SIZE)
+                .put("max_pool_size", rdbSideTableInfo.getAsyncPoolSize())
                 .put("user", rdbSideTableInfo.getUserName())
                 .put("password", rdbSideTableInfo.getPassword())
                 .put("provider_class", DT_PROVIDER_CLASS)
@@ -59,7 +59,7 @@ public void open(Configuration parameters) throws Exception {
         System.setProperty("vertx.disableFileCPResolving", "true");
         VertxOptions vo = new VertxOptions();
         vo.setEventLoopPoolSize(DEFAULT_VERTX_EVENT_LOOP_POOL_SIZE);
-        vo.setWorkerPoolSize(DEFAULT_VERTX_WORKER_POOL_SIZE);
+        vo.setWorkerPoolSize(rdbSideTableInfo.getAsyncPoolSize());
         vo.setFileResolverCachingEnabled(false);
         Vertx vertx = Vertx.vertx(vo);
         setRdbSQLClient(JDBCClient.createNonShared(vertx, clickhouseClientConfig));
diff --git a/core/src/main/java/com/dtstack/flink/sql/side/SideTableInfo.java b/core/src/main/java/com/dtstack/flink/sql/side/SideTableInfo.java
@@ -55,6 +55,8 @@ public abstract class SideTableInfo extends TableInfo implements Serializable {
 
     public static final String ASYNC_TIMEOUT_NUM_KEY = "asyncTimeoutNum";
 
+    public static final String ASYNC_REQ_POOL_KEY = "asyncPoolSize";
+
     private String cacheType = "none";//None or LRU or ALL
 
     private int cacheSize = 10000;
@@ -65,6 +67,11 @@ public abstract class SideTableInfo extends TableInfo implements Serializable {
 
     private int  asyncTimeout=10000;
 
+    /**
+     *  async operator req outside conn pool size, egg rdb conn pool size
+     */
+    private int asyncPoolSize = 0;
+
     private int asyncTimeoutNumLimit = Integer.MAX_VALUE;
 
     private boolean partitionedJoin = false;
@@ -156,4 +163,27 @@ public void setAsyncTimeoutNumLimit(int asyncTimeoutNumLimit) {
         this.asyncTimeoutNumLimit = asyncTimeoutNumLimit;
     }
 
+    public int getAsyncPoolSize() {
+        return asyncPoolSize;
+    }
+
+    public void setAsyncPoolSize(int asyncPoolSize) {
+        this.asyncPoolSize = asyncPoolSize;
+    }
+
+    @Override
+    public String toString() {
+        return "Cache Info{" +
+                "cacheType='" + cacheType + '\'' +
+                ", cacheSize=" + cacheSize +
+                ", cacheTimeout=" + cacheTimeout +
+                ", asyncCapacity=" + asyncCapacity +
+                ", asyncTimeout=" + asyncTimeout +
+                ", asyncPoolSize=" + asyncPoolSize +
+                ", asyncTimeoutNumLimit=" + asyncTimeoutNumLimit +
+                ", partitionedJoin=" + partitionedJoin +
+                ", cacheMode='" + cacheMode + '\'' +
+                '}';
+    }
+
 }
diff --git a/core/src/main/java/com/dtstack/flink/sql/table/AbsSideTableParser.java b/core/src/main/java/com/dtstack/flink/sql/table/AbsSideTableParser.java
@@ -23,6 +23,8 @@
 import com.dtstack.flink.sql.enums.ECacheType;
 import com.dtstack.flink.sql.side.SideTableInfo;
 import com.dtstack.flink.sql.util.MathUtil;
+import org.apache.flink.util.Preconditions;
+
 import java.util.Map;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
@@ -114,6 +116,12 @@ protected void parseCacheProp(SideTableInfo sideTableInfo, Map<String, Object> p
                     sideTableInfo.setAsyncTimeoutNumLimit(asyncTimeoutNum);
                 }
             }
+
+            if (props.containsKey(SideTableInfo.ASYNC_REQ_POOL_KEY.toLowerCase())) {
+                Integer asyncPoolSize = MathUtil.getIntegerVal(props.get(SideTableInfo.ASYNC_REQ_POOL_KEY.toLowerCase()));
+                Preconditions.checkArgument(asyncPoolSize > 0 && asyncPoolSize <= 20, "asyncPoolSize size limit (0,20]");
+                sideTableInfo.setAsyncPoolSize(asyncPoolSize);
+            }
         }
     }
 }
diff --git a/db2/db2-side/db2-async-side/src/main/java/com/dtstack/flink/sql/side/db2/Db2AsyncReqRow.java b/db2/db2-side/db2-async-side/src/main/java/com/dtstack/flink/sql/side/db2/Db2AsyncReqRow.java
@@ -59,7 +59,7 @@ public void open(Configuration parameters) throws Exception {
         RdbSideTableInfo rdbSideTableInfo = (RdbSideTableInfo) sideInfo.getSideTableInfo();
         db2lientConfig.put("url", rdbSideTableInfo.getUrl())
                 .put("driver_class", DB2_DRIVER)
-                .put("max_pool_size", DEFAULT_MAX_DB_CONN_POOL_SIZE)
+                .put("max_pool_size", rdbSideTableInfo.getAsyncPoolSize())
                 .put("user", rdbSideTableInfo.getUserName())
                 .put("password", rdbSideTableInfo.getPassword())
                 .put("provider_class", DT_PROVIDER_CLASS)
@@ -71,7 +71,7 @@ public void open(Configuration parameters) throws Exception {
 
         VertxOptions vo = new VertxOptions();
         vo.setEventLoopPoolSize(DEFAULT_VERTX_EVENT_LOOP_POOL_SIZE);
-        vo.setWorkerPoolSize(DEFAULT_VERTX_WORKER_POOL_SIZE);
+        vo.setWorkerPoolSize(rdbSideTableInfo.getAsyncPoolSize());
         vo.setFileResolverCachingEnabled(false);
         Vertx vertx = Vertx.vertx(vo);
         setRdbSQLClient(JDBCClient.createNonShared(vertx, db2lientConfig));
diff --git a/impala/impala-side/impala-async-side/src/main/java/com/dtstack/flink/sql/side/impala/ImpalaAsyncReqRow.java b/impala/impala-side/impala-async-side/src/main/java/com/dtstack/flink/sql/side/impala/ImpalaAsyncReqRow.java
@@ -57,33 +57,28 @@ public ImpalaAsyncReqRow(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldI
     @Override
     public void open(Configuration parameters) throws Exception {
         super.open(parameters);
-        JsonObject impalaClientConfig = getClientConfig();
+        ImpalaSideTableInfo impalaSideTableInfo = (ImpalaSideTableInfo) sideInfo.getSideTableInfo();
+
+        JsonObject impalaClientConfig = new JsonObject();
+        impalaClientConfig.put("url", getUrl())
+                .put("driver_class", IMPALA_DRIVER)
+                .put("max_pool_size", impalaSideTableInfo.getAsyncPoolSize())
+                .put("provider_class", DT_PROVIDER_CLASS)
+                .put("idle_connection_test_period", 300)
+                .put("test_connection_on_checkin", DEFAULT_TEST_CONNECTION_ON_CHECKIN)
+                .put("max_idle_time", 600);
 
         System.setProperty("vertx.disableFileCPResolving", "true");
 
         VertxOptions vo = new VertxOptions();
         vo.setEventLoopPoolSize(DEFAULT_VERTX_EVENT_LOOP_POOL_SIZE);
-        vo.setWorkerPoolSize(DEFAULT_VERTX_WORKER_POOL_SIZE);
+        vo.setWorkerPoolSize(impalaSideTableInfo.getAsyncPoolSize());
         vo.setFileResolverCachingEnabled(false);
         Vertx vertx = Vertx.vertx(vo);
+
         setRdbSQLClient(JDBCClient.createNonShared(vertx, impalaClientConfig));
     }
 
-    public JsonObject getClientConfig() {
-        JsonObject impalaClientConfig = new JsonObject();
-        impalaClientConfig.put("url", getUrl())
-                .put("driver_class", IMPALA_DRIVER)
-                .put("max_pool_size", DEFAULT_MAX_DB_CONN_POOL_SIZE)
-                .put("provider_class", DT_PROVIDER_CLASS)
-                .put("idle_connection_test_period", 300)
-                .put("test_connection_on_checkin", DEFAULT_TEST_CONNECTION_ON_CHECKIN)
-                .put("max_idle_time", 600)
-                .put("preferred_test_query", PREFERRED_TEST_QUERY_SQL)
-                .put("idle_connection_test_period", DEFAULT_IDLE_CONNECTION_TEST_PEROID)
-                .put("test_connection_on_checkin", DEFAULT_TEST_CONNECTION_ON_CHECKIN);
-
-        return impalaClientConfig;
-    }
 
     public String getUrl() {
         ImpalaSideTableInfo impalaSideTableInfo = (ImpalaSideTableInfo) sideInfo.getSideTableInfo();
diff --git a/mysql/mysql-side/mysql-async-side/src/main/java/com/dtstack/flink/sql/side/mysql/MysqlAsyncReqRow.java b/mysql/mysql-side/mysql-async-side/src/main/java/com/dtstack/flink/sql/side/mysql/MysqlAsyncReqRow.java
@@ -61,7 +61,7 @@ public void open(Configuration parameters) throws Exception {
         RdbSideTableInfo rdbSideTableInfo = (RdbSideTableInfo) sideInfo.getSideTableInfo();
         mysqlClientConfig.put("url", rdbSideTableInfo.getUrl())
                 .put("driver_class", MYSQL_DRIVER)
-                .put("max_pool_size", DEFAULT_MAX_DB_CONN_POOL_SIZE)
+                .put("max_pool_size", rdbSideTableInfo.getAsyncPoolSize())
                 .put("user", rdbSideTableInfo.getUserName())
                 .put("password", rdbSideTableInfo.getPassword())
                 .put("provider_class", DT_PROVIDER_CLASS)
@@ -73,7 +73,7 @@ public void open(Configuration parameters) throws Exception {
 
         VertxOptions vo = new VertxOptions();
         vo.setEventLoopPoolSize(DEFAULT_VERTX_EVENT_LOOP_POOL_SIZE);
-        vo.setWorkerPoolSize(DEFAULT_VERTX_WORKER_POOL_SIZE);
+        vo.setWorkerPoolSize(rdbSideTableInfo.getAsyncPoolSize());
         vo.setFileResolverCachingEnabled(false);
         Vertx vertx = Vertx.vertx(vo);
         setRdbSQLClient(JDBCClient.createNonShared(vertx, mysqlClientConfig));
diff --git a/oracle/oracle-side/oracle-async-side/src/main/java/com/dtstack/flink/sql/side/oracle/OracleAsyncReqRow.java b/oracle/oracle-side/oracle-async-side/src/main/java/com/dtstack/flink/sql/side/oracle/OracleAsyncReqRow.java
@@ -53,7 +53,7 @@ public void open(Configuration parameters) throws Exception {
         RdbSideTableInfo rdbSideTableInfo = (RdbSideTableInfo) sideInfo.getSideTableInfo();
         oracleClientConfig.put("url", rdbSideTableInfo.getUrl())
                 .put("driver_class", ORACLE_DRIVER)
-                .put("max_pool_size", DEFAULT_MAX_DB_CONN_POOL_SIZE)
+                .put("max_pool_size", rdbSideTableInfo.getAsyncPoolSize())
                 .put("user", rdbSideTableInfo.getUserName())
                 .put("password", rdbSideTableInfo.getPassword())
                 .put("provider_class", DT_PROVIDER_CLASS)
@@ -65,7 +65,7 @@ public void open(Configuration parameters) throws Exception {
 
         VertxOptions vo = new VertxOptions();
         vo.setEventLoopPoolSize(DEFAULT_VERTX_EVENT_LOOP_POOL_SIZE);
-        vo.setWorkerPoolSize(DEFAULT_VERTX_WORKER_POOL_SIZE);
+        vo.setWorkerPoolSize(rdbSideTableInfo.getAsyncPoolSize());
         vo.setFileResolverCachingEnabled(false);
         Vertx vertx = Vertx.vertx(vo);
         setRdbSQLClient(JDBCClient.createNonShared(vertx, oracleClientConfig));
diff --git a/polardb/polardb-side/polardb-async-side/src/main/java/com/dtstack/flink/sql/side/polardb/PolardbAsyncReqRow.java b/polardb/polardb-side/polardb-async-side/src/main/java/com/dtstack/flink/sql/side/polardb/PolardbAsyncReqRow.java
@@ -33,7 +33,7 @@ public void open(Configuration parameters) throws Exception {
         RdbSideTableInfo rdbSideTableInfo = (RdbSideTableInfo) sideInfo.getSideTableInfo();
         mysqlClientConfig.put("url", rdbSideTableInfo.getUrl())
                 .put("driver_class", POLARDB_DRIVER)
-                .put("max_pool_size", DEFAULT_MAX_DB_CONN_POOL_SIZE)
+                .put("max_pool_size", rdbSideTableInfo.getAsyncPoolSize())
                 .put("user", rdbSideTableInfo.getUserName())
                 .put("password", rdbSideTableInfo.getPassword())
                 .put("provider_class", DT_PROVIDER_CLASS)
@@ -45,7 +45,7 @@ public void open(Configuration parameters) throws Exception {
 
         VertxOptions vo = new VertxOptions();
         vo.setEventLoopPoolSize(DEFAULT_VERTX_EVENT_LOOP_POOL_SIZE);
-        vo.setWorkerPoolSize(DEFAULT_VERTX_WORKER_POOL_SIZE);
+        vo.setWorkerPoolSize(rdbSideTableInfo.getAsyncPoolSize());
         vo.setFileResolverCachingEnabled(false);
         Vertx vertx = Vertx.vertx(vo);
         setRdbSQLClient(JDBCClient.createNonShared(vertx, mysqlClientConfig));
diff --git a/postgresql/postgresql-side/postgresql-async-side/src/main/java/com/dtstack/flink/sql/side/postgresql/PostgresqlAsyncReqRow.java b/postgresql/postgresql-side/postgresql-async-side/src/main/java/com/dtstack/flink/sql/side/postgresql/PostgresqlAsyncReqRow.java
@@ -60,7 +60,7 @@ public void open(Configuration parameters) throws Exception {
         RdbSideTableInfo rdbSideTableInfo = (RdbSideTableInfo) sideInfo.getSideTableInfo();
         pgClientConfig.put("url", rdbSideTableInfo.getUrl())
                 .put("driver_class", POSTGRESQL_DRIVER)
-                .put("max_pool_size", DEFAULT_MAX_DB_CONN_POOL_SIZE)
+                .put("max_pool_size", rdbSideTableInfo.getAsyncPoolSize())
                 .put("user", rdbSideTableInfo.getUserName())
                 .put("password", rdbSideTableInfo.getPassword())
                 .put("preferred_test_query", PREFERRED_TEST_QUERY_SQL)
@@ -69,7 +69,7 @@ public void open(Configuration parameters) throws Exception {
 
         VertxOptions vo = new VertxOptions();
         vo.setEventLoopPoolSize(DEFAULT_VERTX_EVENT_LOOP_POOL_SIZE);
-        vo.setWorkerPoolSize(DEFAULT_VERTX_WORKER_POOL_SIZE);
+        vo.setWorkerPoolSize(rdbSideTableInfo.getAsyncPoolSize());
         Vertx vertx = Vertx.vertx(vo);
         setRdbSQLClient(JDBCClient.createNonShared(vertx, pgClientConfig));
     }
diff --git a/rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/all/RdbAllReqRow.java b/rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/all/RdbAllReqRow.java
@@ -19,6 +19,7 @@
 package com.dtstack.flink.sql.side.rdb.all;
 
 import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.configuration.Configuration;
 import org.apache.flink.table.runtime.types.CRow;
 import org.apache.flink.table.typeutils.TimeIndicatorTypeInfo;
 import org.apache.flink.types.Row;
@@ -72,6 +73,13 @@ public RdbAllReqRow(SideInfo sideInfo) {
         super(sideInfo);
     }
 
+    @Override
+    public void open(Configuration parameters) throws Exception {
+        super.open(parameters);
+        RdbSideTableInfo tableInfo = (RdbSideTableInfo) sideInfo.getSideTableInfo();
+        LOG.info("rdb dim table config info: {} ", tableInfo.toString());
+    }
+
     @Override
     protected void initCache() throws SQLException {
         Map<String, List<Map<String, Object>>> newCache = Maps.newConcurrentMap();
diff --git a/rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/async/RdbAsyncReqRow.java b/rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/async/RdbAsyncReqRow.java
@@ -22,6 +22,7 @@
 import com.dtstack.flink.sql.enums.ECacheContentType;
 import com.dtstack.flink.sql.side.*;
 import com.dtstack.flink.sql.side.cache.CacheObj;
+import com.dtstack.flink.sql.side.rdb.table.RdbSideTableInfo;
 import com.dtstack.flink.sql.side.rdb.util.SwitchUtil;
 import com.dtstack.flink.sql.util.DateUtil;
 import io.vertx.core.json.JsonArray;
@@ -59,7 +60,9 @@ public class RdbAsyncReqRow extends AsyncReqRow {
 
     public final static int DEFAULT_VERTX_WORKER_POOL_SIZE = Runtime.getRuntime().availableProcessors() * 2;
 
-    public final static int DEFAULT_MAX_DB_CONN_POOL_SIZE = DEFAULT_VERTX_EVENT_LOOP_POOL_SIZE + DEFAULT_VERTX_WORKER_POOL_SIZE;
+    public final static int DEFAULT_DB_CONN_POOL_SIZE = DEFAULT_VERTX_EVENT_LOOP_POOL_SIZE + DEFAULT_VERTX_WORKER_POOL_SIZE;
+
+    public final static int MAX_DB_CONN_POOL_SIZE_LIMIT = 20;
 
     public final static int DEFAULT_IDLE_CONNECTION_TEST_PEROID = 60;
 
@@ -73,6 +76,14 @@ public class RdbAsyncReqRow extends AsyncReqRow {
 
     public RdbAsyncReqRow(SideInfo sideInfo) {
         super(sideInfo);
+        init(sideInfo);
+    }
+
+    protected void init(SideInfo sideInfo) {
+        RdbSideTableInfo rdbSideTableInfo = (RdbSideTableInfo) sideInfo.getSideTableInfo();
+        int defaultAsyncPoolSize = Math.min(MAX_DB_CONN_POOL_SIZE_LIMIT, DEFAULT_DB_CONN_POOL_SIZE);
+        int rdbPoolSize = rdbSideTableInfo.getAsyncPoolSize() > 0 ? rdbSideTableInfo.getAsyncPoolSize() : defaultAsyncPoolSize;
+        rdbSideTableInfo.setAsyncPoolSize(rdbPoolSize);
     }
 
     @Override
diff --git a/rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/table/RdbSideTableInfo.java b/rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/table/RdbSideTableInfo.java
@@ -101,4 +101,16 @@ public String getPassword() {
     public void setPassword(String password) {
         this.password = password;
     }
+
+    @Override
+    public String toString() {
+        String cacheInfo = super.toString();
+        String connectionInfo = "Rdb Side Connection Info{" +
+                "url='" + url + '\'' +
+                ", tableName='" + tableName + '\'' +
+                ", schema='" + schema + '\'' +
+                '}';
+        return cacheInfo + " , " + connectionInfo;
+    }
+
 }
diff --git a/sqlserver/sqlserver-side/sqlserver-async-side/src/main/java/com/dtstack/flink/sql/side/sqlserver/SqlserverAsyncReqRow.java b/sqlserver/sqlserver-side/sqlserver-async-side/src/main/java/com/dtstack/flink/sql/side/sqlserver/SqlserverAsyncReqRow.java
@@ -51,7 +51,7 @@ public void open(Configuration parameters) throws Exception {
         RdbSideTableInfo rdbSideTableInfo = (RdbSideTableInfo) sideInfo.getSideTableInfo();
         sqlserverClientConfig.put("url", rdbSideTableInfo.getUrl())
                 .put("driver_class", SQLSERVER_DRIVER)
-                .put("max_pool_size", DEFAULT_MAX_DB_CONN_POOL_SIZE)
+                .put("max_pool_size", rdbSideTableInfo.getAsyncPoolSize())
                 .put("user", rdbSideTableInfo.getUserName())
                 .put("password", rdbSideTableInfo.getPassword())
                 .put("provider_class", DT_PROVIDER_CLASS)
@@ -64,7 +64,7 @@ public void open(Configuration parameters) throws Exception {
 
         VertxOptions vo = new VertxOptions();
         vo.setEventLoopPoolSize(DEFAULT_VERTX_EVENT_LOOP_POOL_SIZE);
-        vo.setWorkerPoolSize(DEFAULT_VERTX_WORKER_POOL_SIZE);
+        vo.setWorkerPoolSize(rdbSideTableInfo.getAsyncPoolSize());
         vo.setFileResolverCachingEnabled(false);
         Vertx vertx = Vertx.vertx(vo);
         setRdbSQLClient(JDBCClient.createNonShared(vertx, sqlserverClientConfig));

Original file line number	Diff line number	Diff line change
`@@ -23,6 +23,8 @@`
`23`	`23`	`import com.dtstack.flink.sql.enums.ECacheType;`
`24`	`24`	`import com.dtstack.flink.sql.side.SideTableInfo;`
`25`	`25`	`import com.dtstack.flink.sql.util.MathUtil;`
	`26`	`+import org.apache.flink.util.Preconditions;`
	`27`	`+`
`26`	`28`	`import java.util.Map;`
`27`	`29`	`import java.util.regex.Matcher;`
`28`	`30`	`import java.util.regex.Pattern;`
`@@ -114,6 +116,12 @@ protected void parseCacheProp(SideTableInfo sideTableInfo, Map<String, Object> p`
`114`	`116`	`sideTableInfo.setAsyncTimeoutNumLimit(asyncTimeoutNum);`
`115`	`117`	`}`
`116`	`118`	`}`
	`119`	`+`
	`120`	`+ if (props.containsKey(SideTableInfo.ASYNC_REQ_POOL_KEY.toLowerCase())) {`
	`121`	`+ Integer asyncPoolSize = MathUtil.getIntegerVal(props.get(SideTableInfo.ASYNC_REQ_POOL_KEY.toLowerCase()));`
	`122`	`+ Preconditions.checkArgument(asyncPoolSize > 0 && asyncPoolSize <= 20, "asyncPoolSize size limit (0,20]");`
	`123`	`+ sideTableInfo.setAsyncPoolSize(asyncPoolSize);`
	`124`	`+ }`
`117`	`125`	`}`
`118`	`126`	`}`
`119`	`127`	`}`