[improve][ws] Add memory limit configuration for Pulsar client used i…

…n Websocket proxy (apache#22666)

(cherry picked from commit 80d4675)

conf/broker.conf

@@ -1535,6 +1535,9 @@ webSocketNumServiceThreads=
 # Number of connections per Broker in Pulsar Client used in WebSocket proxy
 webSocketConnectionsPerBroker=
 
+# Memory limit in MBs for direct memory in Pulsar Client used in WebSocket proxy
+webSocketPulsarClientMemoryLimitInMB=0
+
 # Time in milliseconds that idle WebSocket session times out
 webSocketSessionIdleTimeoutMillis=300000
 

conf/standalone.conf

@@ -966,6 +966,9 @@ webSocketNumIoThreads=8
 # Number of connections per Broker in Pulsar Client used in WebSocket proxy
 webSocketConnectionsPerBroker=8
 
+# Memory limit in MBs for direct memory in Pulsar Client used in WebSocket proxy
+webSocketPulsarClientMemoryLimitInMB=0
+
 # Time in milliseconds that idle WebSocket session times out
 webSocketSessionIdleTimeoutMillis=300000
 

conf/websocket.conf

@@ -71,6 +71,9 @@ numHttpServerThreads=
 # Number of connections per Broker in Pulsar Client used in WebSocket proxy
 webSocketConnectionsPerBroker=
 
+# Memory limit in MBs for direct memory in Pulsar Client used in WebSocket proxy
+webSocketPulsarClientMemoryLimitInMB=0
+
 # Time in milliseconds that idle WebSocket session times out
 webSocketSessionIdleTimeoutMillis=300000
 

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java

@@ -2836,6 +2836,13 @@ The delayed message index time step(in seconds) in per bucket snapshot segment,
         doc = "Number of connections per Broker in Pulsar Client used in WebSocket proxy"
     )
     private int webSocketConnectionsPerBroker = Runtime.getRuntime().availableProcessors();
+
+    @FieldContext(
+            category = CATEGORY_WEBSOCKET,
+            doc = "Memory limit in MBs for direct memory in Pulsar Client used in WebSocket proxy"
+    )
+    private int webSocketPulsarClientMemoryLimitInMB = 0;
+
     @FieldContext(
         category = CATEGORY_WEBSOCKET,
         doc = "Time in milliseconds that idle WebSocket session times out"

pulsar-websocket/src/main/java/org/apache/pulsar/websocket/WebSocketService.java

@@ -195,7 +195,8 @@ public synchronized void setLocalCluster(ClusterData clusterData) {
 
     private PulsarClient createClientInstance(ClusterData clusterData) throws IOException {
         ClientBuilder clientBuilder = PulsarClient.builder() //
-                .memoryLimit(0, SizeUnit.BYTES)
+                .memoryLimit(SizeUnit.MEGA_BYTES.toBytes(config.getWebSocketPulsarClientMemoryLimitInMB()),
+                        SizeUnit.BYTES)
                 .statsInterval(0, TimeUnit.SECONDS) //
                 .enableTls(config.isTlsEnabled()) //
                 .allowTlsInsecureConnection(config.isTlsAllowInsecureConnection()) //

pulsar-websocket/src/main/java/org/apache/pulsar/websocket/service/WebSocketProxyConfiguration.java

@@ -176,6 +176,9 @@ public class WebSocketProxyConfiguration implements PulsarConfiguration {
     @FieldContext(doc = "Number of connections per broker in Pulsar client used in WebSocket proxy")
     private int webSocketConnectionsPerBroker = Runtime.getRuntime().availableProcessors();
 
+    @FieldContext(doc = "Memory limit in MBs for direct memory in Pulsar Client used in WebSocket proxy")
+    private int webSocketPulsarClientMemoryLimitInMB = 0;
+
     @FieldContext(doc = "Timeout of idling WebSocket session (in milliseconds)")
     private int webSocketSessionIdleTimeoutMillis = 300000;