Remove dn_rpc_min_concurrent_client_num to fix minWorker > maxWorker bugs apache#15296

Signed-off-by: OneSizeFitQuorum <[email protected]>

Author: OneSizeFitsQuorum

iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/config/PipeConsensusConfig.java

@@ -81,39 +81,25 @@ public PipeConsensusConfig build() {
   }
 
   public static class RPC {
-    private final int rpcSelectorThreadNum;
-    private final int rpcMinConcurrentClientNum;
     private final int rpcMaxConcurrentClientNum;
     private final int thriftServerAwaitTimeForStopService;
     private final boolean isRpcThriftCompressionEnabled;
     private final int connectionTimeoutInMs;
     private final int thriftMaxFrameSize;
 
     public RPC(
-        int rpcSelectorThreadNum,
-        int rpcMinConcurrentClientNum,
         int rpcMaxConcurrentClientNum,
         int thriftServerAwaitTimeForStopService,
         boolean isRpcThriftCompressionEnabled,
         int connectionTimeoutInMs,
         int thriftMaxFrameSize) {
-      this.rpcSelectorThreadNum = rpcSelectorThreadNum;
-      this.rpcMinConcurrentClientNum = rpcMinConcurrentClientNum;
       this.rpcMaxConcurrentClientNum = rpcMaxConcurrentClientNum;
       this.thriftServerAwaitTimeForStopService = thriftServerAwaitTimeForStopService;
       this.isRpcThriftCompressionEnabled = isRpcThriftCompressionEnabled;
       this.connectionTimeoutInMs = connectionTimeoutInMs;
       this.thriftMaxFrameSize = thriftMaxFrameSize;
     }
 
-    public int getRpcSelectorThreadNum() {
-      return rpcSelectorThreadNum;
-    }
-
-    public int getRpcMinConcurrentClientNum() {
-      return rpcMinConcurrentClientNum;
-    }
-
     public int getRpcMaxConcurrentClientNum() {
       return rpcMaxConcurrentClientNum;
     }
@@ -139,24 +125,12 @@ public static RPC.Builder newBuilder() {
     }
 
     public static class Builder {
-      private int rpcSelectorThreadNum = 1;
-      private int rpcMinConcurrentClientNum = Runtime.getRuntime().availableProcessors();
       private int rpcMaxConcurrentClientNum = 65535;
       private int thriftServerAwaitTimeForStopService = 60;
       private boolean isRpcThriftCompressionEnabled = false;
       private int connectionTimeoutInMs = (int) TimeUnit.SECONDS.toMillis(60);
       private int thriftMaxFrameSize = 536870912;
 
-      public RPC.Builder setRpcSelectorThreadNum(int rpcSelectorThreadNum) {
-        this.rpcSelectorThreadNum = rpcSelectorThreadNum;
-        return this;
-      }
-
-      public RPC.Builder setRpcMinConcurrentClientNum(int rpcMinConcurrentClientNum) {
-        this.rpcMinConcurrentClientNum = rpcMinConcurrentClientNum;
-        return this;
-      }
-
       public RPC.Builder setRpcMaxConcurrentClientNum(int rpcMaxConcurrentClientNum) {
         this.rpcMaxConcurrentClientNum = rpcMaxConcurrentClientNum;
         return this;
@@ -185,8 +159,6 @@ public RPC.Builder setThriftMaxFrameSize(int thriftMaxFrameSize) {
 
       public RPC build() {
         return new RPC(
-            rpcSelectorThreadNum,
-            rpcMinConcurrentClientNum,
             rpcMaxConcurrentClientNum,
             thriftServerAwaitTimeForStopService,
             isRpcThriftCompressionEnabled,

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java

@@ -145,9 +145,6 @@ public class IoTDBConfig {
   /** Rpc Selector thread num */
   private int rpcSelectorThreadCount = 1;
 
-  /** Min concurrent client number */
-  private int rpcMinConcurrentClientNum = Runtime.getRuntime().availableProcessors();
-
   /** Max concurrent client number */
   private int rpcMaxConcurrentClientNum = 1000;
 
@@ -1770,14 +1767,6 @@ public void setRpcSelectorThreadCount(int rpcSelectorThreadCount) {
     this.rpcSelectorThreadCount = rpcSelectorThreadCount;
   }
 
-  public int getRpcMinConcurrentClientNum() {
-    return rpcMinConcurrentClientNum;
-  }
-
-  public void setRpcMinConcurrentClientNum(int rpcMinConcurrentClientNum) {
-    this.rpcMinConcurrentClientNum = rpcMinConcurrentClientNum;
-  }
-
   public int getRpcMaxConcurrentClientNum() {
     return rpcMaxConcurrentClientNum;
   }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java

@@ -728,18 +728,6 @@ public void loadProperties(TrimProperties properties) throws BadNodeUrlException
 
     conf.setRpcSelectorThreadCount(rpcSelectorThreadNum);
 
-    int minConcurrentClientNum =
-        Integer.parseInt(
-            properties.getProperty(
-                "dn_rpc_min_concurrent_client_num",
-                Integer.toString(conf.getRpcMinConcurrentClientNum())));
-
-    if (minConcurrentClientNum <= 0) {
-      minConcurrentClientNum = Runtime.getRuntime().availableProcessors();
-    }
-
-    conf.setRpcMinConcurrentClientNum(minConcurrentClientNum);
-
     int maxConcurrentClientNum =
         Integer.parseInt(
             properties.getProperty(

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/DataRegionConsensusImpl.java

@@ -129,8 +129,6 @@ private static ConsensusConfig buildConsensusConfig() {
                   .setRpc(
                       RPC.newBuilder()
                           .setConnectionTimeoutInMs(CONF.getConnectionTimeoutInMS())
-                          .setRpcSelectorThreadNum(CONF.getRpcSelectorThreadCount())
-                          .setRpcMinConcurrentClientNum(CONF.getRpcMinConcurrentClientNum())
                           .setRpcMaxConcurrentClientNum(CONF.getRpcMaxConcurrentClientNum())
                           .setRpcThriftCompressionEnabled(CONF.isRpcThriftCompressionEnable())
                           .setSelectorNumOfClientManager(CONF.getSelectorNumOfClientManager())
@@ -157,8 +155,6 @@ private static ConsensusConfig buildConsensusConfig() {
                       PipeConsensusConfig.RPC
                           .newBuilder()
                           .setConnectionTimeoutInMs(CONF.getConnectionTimeoutInMS())
-                          .setRpcSelectorThreadNum(CONF.getRpcSelectorThreadCount())
-                          .setRpcMinConcurrentClientNum(CONF.getRpcMinConcurrentClientNum())
                           .setRpcMaxConcurrentClientNum(CONF.getRpcMaxConcurrentClientNum())
                           .setIsRpcThriftCompressionEnabled(CONF.isRpcThriftCompressionEnable())
                           .setThriftServerAwaitTimeForStopService(

iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template

@@ -502,11 +502,6 @@ dn_rpc_advanced_compression_enable=false
 # Datatype: int
 dn_rpc_selector_thread_count=1
 
-# The min number of concurrent clients that can be connected to the dataNode.
-# effectiveMode: restart
-# Datatype: int
-dn_rpc_min_concurrent_client_num=1
-
 # The maximum number of concurrent clients that can be connected to the dataNode.
 # effectiveMode: restart
 # Datatype: int

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/AbstractThriftServiceThread.java

@@ -113,7 +113,6 @@ protected AbstractThriftServiceThread(
       String bindAddress,
       int port,
       int selectorThreads,
-      int minWorkerThreads,
       int maxWorkerThreads,
       int timeoutSecond,
       TServerEventHandler serverEventHandler,
@@ -134,7 +133,6 @@ protected AbstractThriftServiceThread(
                   processor,
                   threadsName,
                   selectorThreads,
-                  minWorkerThreads,
                   maxWorkerThreads,
                   timeoutSecond,
                   maxReadBufferBytes);
@@ -143,12 +141,7 @@ protected AbstractThriftServiceThread(
         case HSHA:
           THsHaServer.Args poolArgs1 =
               initAsyncedHshaPoolArgs(
-                  processor,
-                  threadsName,
-                  minWorkerThreads,
-                  maxWorkerThreads,
-                  timeoutSecond,
-                  maxReadBufferBytes);
+                  processor, threadsName, maxWorkerThreads, timeoutSecond, maxReadBufferBytes);
           poolServer = new THsHaServer(poolArgs1);
           break;
         default:
@@ -228,10 +221,7 @@ protected AbstractThriftServiceThread(
   private TThreadPoolServer.Args initSyncedPoolArgs(
       TProcessor processor, String threadsName, int maxWorkerThreads, int timeoutSecond) {
     TThreadPoolServer.Args poolArgs = new TThreadPoolServer.Args(serverTransport);
-    poolArgs
-        .maxWorkerThreads(maxWorkerThreads)
-        .minWorkerThreads(Runtime.getRuntime().availableProcessors())
-        .stopTimeoutVal(timeoutSecond);
+    poolArgs.maxWorkerThreads(maxWorkerThreads).minWorkerThreads(0).stopTimeoutVal(timeoutSecond);
     executorService = IoTDBThreadPoolFactory.createThriftRpcClientThreadPool(poolArgs, threadsName);
     poolArgs.executorService = executorService;
     poolArgs.processor(processor);
@@ -244,7 +234,6 @@ private TThreadedSelectorServer.Args initAsyncedSelectorPoolArgs(
       TBaseAsyncProcessor<?> processor,
       String threadsName,
       int selectorThreads,
-      int minWorkerThreads,
       int maxWorkerThreads,
       int timeoutSecond,
       int maxReadBufferBytes) {
@@ -254,7 +243,7 @@ private TThreadedSelectorServer.Args initAsyncedSelectorPoolArgs(
     poolArgs.selectorThreads(selectorThreads);
     executorService =
         IoTDBThreadPoolFactory.createThriftRpcClientThreadPool(
-            minWorkerThreads, maxWorkerThreads, timeoutSecond, TimeUnit.SECONDS, threadsName);
+            0, maxWorkerThreads, timeoutSecond, TimeUnit.SECONDS, threadsName);
     poolArgs.executorService(executorService);
     poolArgs.processor(processor);
     poolArgs.protocolFactory(protocolFactory);
@@ -265,15 +254,14 @@ private TThreadedSelectorServer.Args initAsyncedSelectorPoolArgs(
   private THsHaServer.Args initAsyncedHshaPoolArgs(
       TBaseAsyncProcessor<?> processor,
       String threadsName,
-      int minWorkerThreads,
       int maxWorkerThreads,
       int timeoutSecond,
       int maxReadBufferBytes) {
     THsHaServer.Args poolArgs = new THsHaServer.Args((TNonblockingServerTransport) serverTransport);
     poolArgs.maxReadBufferBytes = maxReadBufferBytes;
     executorService =
         IoTDBThreadPoolFactory.createThriftRpcClientThreadPool(
-            minWorkerThreads, maxWorkerThreads, timeoutSecond, TimeUnit.SECONDS, threadsName);
+            0, maxWorkerThreads, timeoutSecond, TimeUnit.SECONDS, threadsName);
     poolArgs.executorService(executorService);
     poolArgs.processor(processor);
     poolArgs.protocolFactory(protocolFactory);

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/ThriftServiceThread.java

@@ -35,7 +35,6 @@ public ThriftServiceThread(
       String bindAddress,
       int port,
       int selectorThreads,
-      int minWorkerThreads,
       int maxWorkerThreads,
       int timeoutSecond,
       TServerEventHandler serverEventHandler,
@@ -51,7 +50,6 @@ public ThriftServiceThread(
         bindAddress,
         port,
         selectorThreads,
-        minWorkerThreads,
         maxWorkerThreads,
         timeoutSecond,
         serverEventHandler,