Parameterize EventHub threadpool configurations

wso2 · Oct 4, 2023 · 55ccfbc · 55ccfbc
1 parent a0e8dbd
commit 55ccfbc
Show file tree

Hide file tree

Showing 11 changed files with 106 additions and 23 deletions.
diff --git a/.../src/main/java/org/wso2/carbon/apimgt/cache/invalidation/APIMgtServerStartupListener.java b/.../src/main/java/org/wso2/carbon/apimgt/cache/invalidation/APIMgtServerStartupListener.java
@@ -35,12 +35,11 @@ public class APIMgtServerStartupListener implements ServerStartupObserver, Serve
 
     public APIMgtServerStartupListener() {
 
-        EventHubConfigurationDto.EventHubReceiverConfiguration eventHubReceiverConfiguration =
+        EventHubConfigurationDto eventHubConfiguration =
                 DataHolder.getInstance().getAPIManagerConfigurationService().getAPIManagerConfiguration()
-                        .getEventHubConfigurationDto().getEventHubReceiverConfiguration();
-        if (eventHubReceiverConfiguration != null) {
-            this.jmsTransportHandlerForEventHub =
-                    new JMSTransportHandler(eventHubReceiverConfiguration.getJmsConnectionParameters());
+                        .getEventHubConfigurationDto();
+        if (eventHubConfiguration.getEventHubReceiverConfiguration() != null) {
+            this.jmsTransportHandlerForEventHub = new JMSTransportHandler(eventHubConfiguration);
         }
     }
 

diff --git a/components/apimgt/org.wso2.carbon.apimgt.common.jms/pom.xml b/components/apimgt/org.wso2.carbon.apimgt.common.jms/pom.xml
@@ -34,6 +34,10 @@
             <classifier>runtime</classifier>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.wso2.carbon.apimgt</groupId>
+            <artifactId>org.wso2.carbon.apimgt.impl</artifactId>
+        </dependency>
     </dependencies>
 
     <build>

diff --git a/...pimgt.common.jms/src/main/java/org/wso2/carbon/apimgt/common/jms/JMSTransportHandler.java b/...pimgt.common.jms/src/main/java/org/wso2/carbon/apimgt/common/jms/JMSTransportHandler.java
@@ -23,6 +23,7 @@
 import org.apache.commons.logging.LogFactory;
 import org.wso2.carbon.apimgt.common.jms.factory.JMSConnectionFactory;
 import org.wso2.carbon.apimgt.common.jms.factory.JMSTaskManagerFactory;
+import org.wso2.carbon.apimgt.impl.dto.EventHubConfigurationDto;
 
 import java.io.IOException;
 import java.io.InputStream;
@@ -73,6 +74,40 @@ public JMSTransportHandler(Properties jmsConnectionProperties) {
         jmsConnectionFactory = new JMSConnectionFactory(parameters, ListenerConstants.CONNECTION_FACTORY_NAME);
     }
 
+    /**
+     * Initializes a JMSTransportHandler with the provided EventHubConfigurationDto.
+     *
+     * @param eventHubConfiguration The EventHubConfigurationDto containing configuration parameters
+     *                              for the JMSTransportHandler.
+     */
+    public JMSTransportHandler(EventHubConfigurationDto eventHubConfiguration) {
+
+        Properties jmsConnectionProperties =
+                eventHubConfiguration.getEventHubReceiverConfiguration().getJmsConnectionParameters();
+        Properties properties;
+        Hashtable<String, String> parameters = new Hashtable<>();
+
+        if (jmsConnectionProperties.isEmpty()) {
+            properties = new Properties();
+            ClassLoader classLoader = getClass().getClassLoader();
+            try (InputStream resourceStream = classLoader.getResourceAsStream(ListenerConstants.MB_PROPERTIES)) {
+                properties.load(resourceStream);
+            } catch (IOException e) {
+                log.error("Cannot read properties file from resources. " + e.getMessage(), e);
+            }
+        } else {
+            properties = jmsConnectionProperties;
+        }
+
+        for (final String name : properties.stringPropertyNames()) {
+            parameters.put(name, properties.getProperty(name));
+        }
+        jmsConnectionFactory = new JMSConnectionFactory(parameters, ListenerConstants.CONNECTION_FACTORY_NAME);
+        minThreadPoolSize = eventHubConfiguration.getMinThreadPoolSize();
+        maxThreadPoolSize = eventHubConfiguration.getMaxThreadPoolSize();
+        jobQueueSize = eventHubConfiguration.getThreadPoolQueueLength();
+    }
+
     /**
      * This method is used to subscribe to JMS topics and receive JMS messages
      * @param messageListener

diff --git a/...ateway/src/main/java/org/wso2/carbon/apimgt/gateway/listeners/GatewayStartupListener.java b/...ateway/src/main/java/org/wso2/carbon/apimgt/gateway/listeners/GatewayStartupListener.java
@@ -100,12 +100,10 @@ public GatewayStartupListener() {
                 throttleProperties.getJmsConnectionProperties();
         this.jmsTransportHandlerForTrafficManager =
                 new JMSTransportHandler(jmsConnectionProperties.getJmsConnectionProperties());
-        EventHubConfigurationDto.EventHubReceiverConfiguration eventHubReceiverConfiguration =
-                ServiceReferenceHolder.getInstance().getAPIManagerConfiguration().getEventHubConfigurationDto()
-                        .getEventHubReceiverConfiguration();
-        if (eventHubReceiverConfiguration != null) {
-            this.jmsTransportHandlerForEventHub =
-                    new JMSTransportHandler(eventHubReceiverConfiguration.getJmsConnectionParameters());
+        EventHubConfigurationDto eventHubConfiguration =
+                ServiceReferenceHolder.getInstance().getAPIManagerConfiguration().getEventHubConfigurationDto();
+        if (eventHubConfiguration.getEventHubReceiverConfiguration() != null) {
+            this.jmsTransportHandlerForEventHub = new JMSTransportHandler(eventHubConfiguration);
         }
     }
 

diff --git a/...t/org.wso2.carbon.apimgt.impl/src/main/java/org/wso2/carbon/apimgt/impl/APIConstants.java b/...t/org.wso2.carbon.apimgt.impl/src/main/java/org/wso2/carbon/apimgt/impl/APIConstants.java
@@ -2513,6 +2513,9 @@ public static class KeyManager {
 
         public static final String SERVICE_URL = "ServiceURL";
         public static final String INIT_DELAY = "InitDelay";
+        public static final String MIN_THREAD_POOL_SIZE = "MinThreadPoolSize";
+        public static final String MAX_THREAD_POOL_SIZE = "MaxThreadPoolSize";
+        public static final String THREAD_POOL_QUEUE_LENGTH = "ThreadPoolQueueLength";
         public static final String INTROSPECTION_ENDPOINT = "introspection_endpoint";
         public static final String CLIENT_REGISTRATION_ENDPOINT = "client_registration_endpoint";
         public static final String KEY_MANAGER_OPERATIONS_DCR_ENDPOINT = "/keymanager-operations/dcr/register";

diff --git a/...carbon.apimgt.impl/src/main/java/org/wso2/carbon/apimgt/impl/APIManagerConfiguration.java b/...carbon.apimgt.impl/src/main/java/org/wso2/carbon/apimgt/impl/APIManagerConfiguration.java
@@ -1971,6 +1971,21 @@ private void setEventHubConfiguration(OMElement omElement) {
             if (initDelay != null) {
                 eventHubConfigurationDto.setInitDelay(Integer.parseInt(initDelay.getText()));
             }
+            OMElement minThreadPoolSize = omElement.getFirstChildWithName(
+                    new QName(APIConstants.KeyManager.MIN_THREAD_POOL_SIZE));
+            if (minThreadPoolSize != null) {
+                eventHubConfigurationDto.setMinThreadPoolSize(Integer.parseInt(minThreadPoolSize.getText()));
+            }
+            OMElement maxThreadPoolSize = omElement.getFirstChildWithName(
+                    new QName(APIConstants.KeyManager.MAX_THREAD_POOL_SIZE));
+            if (maxThreadPoolSize != null) {
+                eventHubConfigurationDto.setMaxThreadPoolSize(Integer.parseInt(maxThreadPoolSize.getText()));
+            }
+            OMElement threadPoolQueueLength = omElement.getFirstChildWithName(
+                    new QName(APIConstants.KeyManager.THREAD_POOL_QUEUE_LENGTH));
+            if (threadPoolQueueLength != null) {
+                eventHubConfigurationDto.setThreadPoolQueueLength(Integer.parseInt(threadPoolQueueLength.getText()));
+            }
             OMElement usernameElement = omElement.getFirstChildWithName(new QName(APIConstants.KeyManager.USERNAME));
             if (usernameElement != null) {
                 eventHubConfigurationDto.setUsername(usernameElement.getText());

diff --git a/...n.apimgt.impl/src/main/java/org/wso2/carbon/apimgt/impl/dto/EventHubConfigurationDto.java b/...n.apimgt.impl/src/main/java/org/wso2/carbon/apimgt/impl/dto/EventHubConfigurationDto.java
@@ -24,6 +24,9 @@ public class EventHubConfigurationDto {
     private boolean enabled = false;
     private String serviceUrl;
     private int initDelay = 1000;
+    private int minThreadPoolSize;
+    private int maxThreadPoolSize;
+    private int threadPoolQueueLength;
     private String username;
     private char[] password;
     private EventHubReceiverConfiguration eventHubReceiverConfiguration;
@@ -60,6 +63,30 @@ public void setInitDelay(int initDelay) {
         this.initDelay = initDelay;
     }
 
+    public int getMinThreadPoolSize() {
+        return minThreadPoolSize;
+    }
+
+    public void setMinThreadPoolSize(int minThreadPoolSize) {
+        this.minThreadPoolSize = minThreadPoolSize;
+    }
+
+    public int getMaxThreadPoolSize() {
+        return maxThreadPoolSize;
+    }
+
+    public void setMaxThreadPoolSize(int maxThreadPoolSize) {
+        this.maxThreadPoolSize = maxThreadPoolSize;
+    }
+
+    public int getThreadPoolQueueLength() {
+        return threadPoolQueueLength;
+    }
+
+    public void setThreadPoolQueueLength(int threadPoolQueueLength) {
+        this.threadPoolQueueLength = threadPoolQueueLength;
+    }
+
     public EventHubReceiverConfiguration getEventHubReceiverConfiguration() {
 
         return eventHubReceiverConfiguration;

diff --git a/...in/java/org/wso2/carbon/apimgt/jms/listener/utils/JMSListenerStartupShutdownListener.java b/...in/java/org/wso2/carbon/apimgt/jms/listener/utils/JMSListenerStartupShutdownListener.java
@@ -41,12 +41,10 @@ public class JMSListenerStartupShutdownListener implements ServerStartupObserver
 
     public JMSListenerStartupShutdownListener() {
 
-        EventHubConfigurationDto.EventHubReceiverConfiguration eventHubReceiverConfiguration =
-                ServiceReferenceHolder.getInstance().getAPIMConfiguration().getEventHubConfigurationDto()
-                        .getEventHubReceiverConfiguration();
-        if (eventHubReceiverConfiguration != null) {
-            this.jmsTransportHandlerForEventHub =
-                    new JMSTransportHandler(eventHubReceiverConfiguration.getJmsConnectionParameters());
+        EventHubConfigurationDto eventHubConfiguration =
+                ServiceReferenceHolder.getInstance().getAPIMConfiguration().getEventHubConfigurationDto();
+        if (eventHubConfiguration.getEventHubReceiverConfiguration() != null) {
+            this.jmsTransportHandlerForEventHub = new JMSTransportHandler(eventHubConfiguration);
         }
 
     }

diff --git a/.../org/wso2/carbon/apimgt/throttle/policy/deployer/utils/ThrottlePolicyStartupListener.java b/.../org/wso2/carbon/apimgt/throttle/policy/deployer/utils/ThrottlePolicyStartupListener.java
@@ -38,12 +38,10 @@ public class ThrottlePolicyStartupListener implements ServerStartupObserver, Ser
 
     public ThrottlePolicyStartupListener() {
 
-        EventHubConfigurationDto.EventHubReceiverConfiguration eventHubReceiverConfiguration =
-                ServiceReferenceHolder.getInstance().getAPIMConfiguration().getEventHubConfigurationDto()
-                        .getEventHubReceiverConfiguration();
-        if (eventHubReceiverConfiguration != null) {
-            this.jmsTransportHandlerForEventHub =
-                    new JMSTransportHandler(eventHubReceiverConfiguration.getJmsConnectionParameters());
+        EventHubConfigurationDto eventHubConfiguration =
+                ServiceReferenceHolder.getInstance().getAPIMConfiguration().getEventHubConfigurationDto();
+        if (eventHubConfiguration.getEventHubReceiverConfiguration() != null) {
+            this.jmsTransportHandlerForEventHub = new JMSTransportHandler(eventHubConfiguration);
         }
     }
 

diff --git a/...t.core.feature/src/main/resources/conf_templates/org.wso2.carbon.apimgt.core.default.json b/...t.core.feature/src/main/resources/conf_templates/org.wso2.carbon.apimgt.core.default.json
@@ -166,6 +166,9 @@
   "apim.key_manager.enable_retriever": true,
   "apim.event_hub.username": "$ref{apim.throttling.username}",
   "apim.event_hub.password": "$ref{apim.throttling.password}",
+  "apim.event_hub.min_thread_pool_size": "20",
+  "apim.event_hub.max_thread_pool_size": "100",
+  "apim.event_hub.thread_pool_queue_length": "10",
   "apim.key_manager.type": "default",
   "apim.devportal.enable_cross_tenant_subscriptions": false,
   "apim.devportal.default_reserved_username": "apim_reserved_user",

diff --git a/...re.feature/src/main/resources/conf_templates/templates/repository/conf/api-manager.xml.j2 b/...re.feature/src/main/resources/conf_templates/templates/repository/conf/api-manager.xml.j2
@@ -1301,6 +1301,9 @@
          {% if apim.event_hub.init_delay is defined %}
         <InitDelay>{{apim.event_hub.init_delay}}</InitDelay>
           {% endif %}
+        <MinThreadPoolSize>{{apim.event_hub.min_thread_pool_size}}</MinThreadPoolSize>
+        <MaxThreadPoolSize>{{apim.event_hub.max_thread_pool_size}}</MaxThreadPoolSize>
+        <ThreadPoolQueueLength>{{apim.event_hub.thread_pool_queue_length}}</ThreadPoolQueueLength>
         <EventPublisherConfiguration>
             <Type>{{apim.event_hub.event_type}}</Type>
             {% if apim.event_hub.publish.url_group is defined %}