Skip to content

Commit c0a56ef

Browse files
committed
修复异步维表异常
1 parent 4e747f5 commit c0a56ef

File tree

1 file changed

+6
-0
lines changed

1 file changed

+6
-0
lines changed

tidb/tidb-side/tidb-async-side/src/main/java/com/dtstack/flink/sql/side/tidb/TidbAsyncReqRow.java

+6
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@
1818

1919
package com.dtstack.flink.sql.side.tidb;
2020

21+
import com.dtstack.flink.sql.factory.DTThreadFactory;
2122
import com.dtstack.flink.sql.side.AbstractSideTableInfo;
2223
import com.dtstack.flink.sql.side.FieldInfo;
2324
import com.dtstack.flink.sql.side.JoinInfo;
@@ -31,6 +32,9 @@
3132
import org.apache.flink.configuration.Configuration;
3233

3334
import java.util.List;
35+
import java.util.concurrent.LinkedBlockingQueue;
36+
import java.util.concurrent.ThreadPoolExecutor;
37+
import java.util.concurrent.TimeUnit;
3438

3539
/**
3640
* @author tiezhu
@@ -68,5 +72,7 @@ public void open(Configuration parameters) throws Exception {
6872
vo.setFileResolverCachingEnabled(false);
6973
Vertx vertx = Vertx.vertx(vo);
7074
setRdbSqlClient(JDBCClient.createNonShared(vertx, tidbClientConfig));
75+
setExecutor(new ThreadPoolExecutor(50, 50, 0, TimeUnit.MILLISECONDS,
76+
new LinkedBlockingQueue<>(10000), new DTThreadFactory("TidbAsyncExec"), new ThreadPoolExecutor.CallerRunsPolicy()));
7177
}
7278
}

0 commit comments

Comments
 (0)