Make the proxy adapter worker thread configrable (#1549)

Author: Technoboy-

mqtt-proxy/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)"

mqtt-proxy/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())