BP: [apache#2354] feat(client): Explicitly setting grpc netty based event loop threads to avoid too much threads apache#2355

Author: Junfan Zhang

common/src/main/java/org/apache/uniffle/common/config/RssClientConf.java

@@ -219,4 +219,10 @@ public class RssClientConf {
           .asList()
           .noDefaultValue()
           .withDescription("the extra java properties could be configured by this option");
+
+  public static final ConfigOption<Integer> RSS_CLIENT_GRPC_EVENT_LOOP_THREADS =
+          ConfigOptions.key("rss.client.grpc.nettyEventLoopThreads")
+                  .intType()
+                  .defaultValue(-1)
+                  .withDescription("the event loop threads of netty impl for grpc");
 }

internal-client/src/main/java/org/apache/uniffle/client/impl/grpc/CoordinatorGrpcClient.java

@@ -90,7 +90,7 @@ public CoordinatorGrpcClient(String host, int port, int maxRetryAttempts) {
   }
 
   public CoordinatorGrpcClient(String host, int port, int maxRetryAttempts, boolean usePlaintext) {
-    super(host, port, maxRetryAttempts, usePlaintext);
+    super(host, port, maxRetryAttempts, usePlaintext, -1);
     blockingStub = CoordinatorServerGrpc.newBlockingStub(channel);
     LOG.info(
         "Created CoordinatorGrpcClient, host:{}, port:{}, maxRetryAttempts:{}, usePlaintext:{}",

internal-client/src/main/java/org/apache/uniffle/client/impl/grpc/GrpcClient.java

@@ -33,12 +33,17 @@ public abstract class GrpcClient {
   protected int maxRetryAttempts;
   protected ManagedChannel channel;
 
-  protected GrpcClient(String host, int port, int maxRetryAttempts, boolean usePlaintext) {
+  protected GrpcClient(String host, int port, int maxRetryAttempts, boolean usePlaintext, int nettyEventLoopThreads) {
     this.host = host;
     this.port = port;
     this.maxRetryAttempts = maxRetryAttempts;
     this.usePlaintext = usePlaintext;
 
+    if (nettyEventLoopThreads > 0) {
+      System.setProperty(
+              "io.grpc.netty.shaded.io.netty.eventLoopThreads", String.valueOf(nettyEventLoopThreads));
+    }
+
     // build channel
     ManagedChannelBuilder<?> channelBuilder = ManagedChannelBuilder.forAddress(host, port);
 

internal-client/src/main/java/org/apache/uniffle/client/impl/grpc/ShuffleManagerGrpcClient.java

@@ -62,7 +62,7 @@ public ShuffleManagerGrpcClient(String host, int port, int maxRetryAttempts) {
 
   public ShuffleManagerGrpcClient(
       String host, int port, int maxRetryAttempts, boolean usePlaintext) {
-    super(host, port, maxRetryAttempts, usePlaintext);
+    super(host, port, maxRetryAttempts, usePlaintext, -1);
     blockingStub = ShuffleManagerGrpc.newBlockingStub(channel);
   }
 

internal-client/src/main/java/org/apache/uniffle/client/impl/grpc/ShuffleServerGrpcClient.java

@@ -110,6 +110,7 @@
 import org.apache.uniffle.proto.ShuffleServerGrpc;
 import org.apache.uniffle.proto.ShuffleServerGrpc.ShuffleServerBlockingStub;
 
+import static org.apache.uniffle.common.config.RssClientConf.RSS_CLIENT_GRPC_EVENT_LOOP_THREADS;
 import static org.apache.uniffle.proto.RssProtos.StatusCode.NO_BUFFER;
 
 public class ShuffleServerGrpcClient extends GrpcClient implements ShuffleServerClient {
@@ -149,16 +150,18 @@ public ShuffleServerGrpcClient(RssConf rssConf, String host, int port) {
             : rssConf.getInteger(RssClientConf.RPC_MAX_ATTEMPTS),
         rssConf == null
             ? RssClientConf.RPC_TIMEOUT_MS.defaultValue()
-            : rssConf.getLong(RssClientConf.RPC_TIMEOUT_MS));
+            : rssConf.getLong(RssClientConf.RPC_TIMEOUT_MS),
+            true,
+            rssConf.get(RSS_CLIENT_GRPC_EVENT_LOOP_THREADS));
   }
 
   public ShuffleServerGrpcClient(String host, int port, int maxRetryAttempts, long rpcTimeoutMs) {
-    this(host, port, maxRetryAttempts, rpcTimeoutMs, true);
+    this(host, port, maxRetryAttempts, rpcTimeoutMs, true, -1);
   }
 
   public ShuffleServerGrpcClient(
-      String host, int port, int maxRetryAttempts, long rpcTimeoutMs, boolean usePlaintext) {
-    super(host, port, maxRetryAttempts, usePlaintext);
+      String host, int port, int maxRetryAttempts, long rpcTimeoutMs, boolean usePlaintext, int nettyEventLoopThreads) {
+    super(host, port, maxRetryAttempts, usePlaintext, nettyEventLoopThreads);
     blockingStub = ShuffleServerGrpc.newBlockingStub(channel);
     rpcTimeout = rpcTimeoutMs;
   }

internal-client/src/main/java/org/apache/uniffle/client/impl/grpc/ShuffleServerInternalGrpcClient.java

@@ -49,7 +49,7 @@ public ShuffleServerInternalGrpcClient(String host, int port, int maxRetryAttemp
 
   public ShuffleServerInternalGrpcClient(
       String host, int port, int maxRetryAttempts, boolean usePlaintext) {
-    super(host, port, maxRetryAttempts, usePlaintext);
+    super(host, port, maxRetryAttempts, usePlaintext, -1);
     // todo Add ClientInterceptor for authentication
     blockingStub = ShuffleServerInternalGrpc.newBlockingStub(channel);
   }