Make the proxy adapter worker thread configrable (#1549)

streamnative · Nov 27, 2024 · 4764775 · 4764775
1 parent 7d76b42
commit 4764775
Show file tree

Hide file tree

Showing 2 changed files with 9 additions and 2 deletions.
diff --git a/...roxy/src/main/java/io/streamnative/pulsar/handlers/mqtt/proxy/MQTTProxyConfiguration.java b/...roxy/src/main/java/io/streamnative/pulsar/handlers/mqtt/proxy/MQTTProxyConfiguration.java
@@ -69,6 +69,13 @@ public class MQTTProxyConfiguration extends MQTTCommonConfiguration {
     )
     private int maxNoOfChannels = 1;
 
+    @FieldContext(
+            category = CATEGORY_MQTT,
+            doc = "Number of threads to use for Netty IO for proxy to broker. "
+                    + "Default is set to `Runtime.getRuntime().availableProcessors()`"
+    )
+    private int proxyAdapterNumIOThreads = Runtime.getRuntime().availableProcessors();
+
     @FieldContext(
             category = CATEGORY_MQTT,
             doc = "Proxy connect to broker timeout (ms)"

diff --git a/...xy/src/main/java/io/streamnative/pulsar/handlers/mqtt/proxy/channel/MQTTProxyAdapter.java b/...xy/src/main/java/io/streamnative/pulsar/handlers/mqtt/proxy/channel/MQTTProxyAdapter.java
@@ -71,14 +71,14 @@ public class MQTTProxyAdapter {
     private final AtomicLong counter = new AtomicLong(0);
     @Getter
     private final ConcurrentMap<InetSocketAddress, Map<Integer, CompletableFuture<Channel>>> pool;
-    private final int workerThread = Runtime.getRuntime().availableProcessors();
     private final int maxNoOfChannels;
 
     public MQTTProxyAdapter(MQTTProxyService proxyService) {
         this.proxyService = proxyService;
         this.pool = new ConcurrentHashMap<>();
         this.bootstrap = new Bootstrap();
-        this.eventLoopGroup = EventLoopUtil.newEventLoopGroup(workerThread, false, threadFactory);
+        this.eventLoopGroup = EventLoopUtil.newEventLoopGroup(
+                proxyService.getProxyConfig().getProxyAdapterNumIOThreads(), false, threadFactory);
         this.maxNoOfChannels = proxyService.getProxyConfig().getMaxNoOfChannels();
         this.bootstrap.group(eventLoopGroup)
                 .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, proxyService.getProxyConfig().getConnectTimeoutMs())