Merge pull request #857 from nisan-abeywickrama/5.3.x

[5.3.x] Replace hardcoded JMS connection pool configurations with TOML-based configuration support

Author: tharikaGitHub

components/event-publisher/event-output-adapters/org.wso2.carbon.event.output.adapter.jms/src/main/java/org/wso2/carbon/event/output/adapter/jms/JMSEventAdapter.java

@@ -114,7 +114,7 @@ public void testConnect() throws TestConnectionNotSupportedException {
         try {
             Hashtable<String, String> adaptorProperties = new Hashtable<String, String>();
             adaptorProperties.putAll(eventAdapterConfiguration.getStaticProperties());
-            JMSConnectionFactory jmsConnectionFactory = new JMSConnectionFactory(adaptorProperties, eventAdapterConfiguration.getName(), adaptorProperties.get(JMSEventAdapterConstants.ADAPTER_JMS_DESTINATION), 1);
+            JMSConnectionFactory jmsConnectionFactory = new JMSConnectionFactory(adaptorProperties, eventAdapterConfiguration.getName(), adaptorProperties.get(JMSEventAdapterConstants.ADAPTER_JMS_DESTINATION), 1, globalProperties);
             Connection connection = jmsConnectionFactory.createConnection();
             connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
             connection.close();
@@ -199,7 +199,7 @@ private PublisherDetails initPublisher(
             maxConnections = JMSEventAdapterConstants.ADAPTER_MAX_THREAD_POOL_SIZE;
         }
 
-        JMSConnectionFactory jmsConnectionFactory = new JMSConnectionFactory(adapterProperties, outputEventAdaptorConfiguration.getName(), messageConfig.get(JMSEventAdapterConstants.ADAPTER_JMS_DESTINATION), maxConnections);
+        JMSConnectionFactory jmsConnectionFactory = new JMSConnectionFactory(adapterProperties, outputEventAdaptorConfiguration.getName(), messageConfig.get(JMSEventAdapterConstants.ADAPTER_JMS_DESTINATION), maxConnections, globalProperties);
         JMSMessageSender jmsMessageSender = new JMSMessageSender(jmsConnectionFactory);
         publisherDetails = new PublisherDetails(jmsConnectionFactory, jmsMessageSender, messageConfig);
 

components/event-publisher/event-output-adapters/org.wso2.carbon.event.output.adapter.jms/src/main/java/org/wso2/carbon/event/output/adapter/jms/internal/util/JMSConnectionFactory.java

@@ -28,6 +28,7 @@
 import javax.naming.InitialContext;
 import javax.naming.NamingException;
 import java.util.Hashtable;
+import java.util.Map;
 
 /**
  * Encapsulate a JMS Connection factory definition within an Axis2.xml
@@ -64,17 +65,20 @@ public class JMSConnectionFactory {
 
     private int maxConnections;
 
+    private Map<String, String> jmsConnectionProperties;
+
     private GenericObjectPool connectionPool;
 
     private String destinationName;
 
     /**
      * Digest a JMS CF definition from an axis2.xml 'Parameter' and construct
      */
-    public JMSConnectionFactory(Hashtable<String, String> parameters, String name, String destination, int maxConcurrentConnections) {
+    public JMSConnectionFactory(Hashtable<String, String> parameters, String name, String destination, int maxConcurrentConnections, Map<String, String> jmsConnectionProperties) {
         this.parameters = parameters;
         this.name = name;
         this.destinationName = destination;
+        this.jmsConnectionProperties = jmsConnectionProperties;
 
         if (maxConcurrentConnections > 0) {
             this.maxConnections = maxConcurrentConnections;
@@ -100,13 +104,13 @@ public JMSConnectionFactory(Hashtable<String, String> parameters, String name, S
     // need to initialize
     private void createConnectionPool() {
         GenericObjectPool.Config poolConfig = new GenericObjectPool.Config();
-        poolConfig.minEvictableIdleTimeMillis = 3000;
-        poolConfig.maxWait = 3000;
+        poolConfig.minEvictableIdleTimeMillis = (Long) parseJmsProperty(JMSEventAdapterConstants.ADAPTER_MIN_EVICTABLE_IDLE_TIME_NAME, 3000L);
+        poolConfig.maxWait = (Long) parseJmsProperty(JMSEventAdapterConstants.ADAPTER_MAX_WAIT_NAME, 3000L);
         poolConfig.maxActive = maxConnections;
         poolConfig.maxIdle = maxConnections;
-        poolConfig.minIdle = 0;
+        poolConfig.minIdle = (Integer) parseJmsProperty(JMSEventAdapterConstants.ADAPTER_MIN_IDLE_NAME, 0);
         poolConfig.numTestsPerEvictionRun = Math.max(1, maxConnections / 10);
-        poolConfig.timeBetweenEvictionRunsMillis = 5000;
+        poolConfig.timeBetweenEvictionRunsMillis = (Long) parseJmsProperty(JMSEventAdapterConstants.ADAPTER_TIME_BETWEEN_EVICTION_RUNS_NAME, 5000L);
         this.connectionPool = new GenericObjectPool(new PoolableJMSConnectionFactory(), poolConfig);
 
     }
@@ -289,6 +293,25 @@ public JMSPooledConnectionHolder getConnectionFromPool() {
         }
     }
 
+    /**
+     * Parses a JMS property value and returns it as the specified type (Integer or Long).
+     * If the value is empty, returns the provided default value.
+     *
+     * @param jmsPropertyKey the JMS property key as a String
+     * @param defaultValue the default value to use if the value is empty
+     * @return the parsed value as Integer or Long, based on the type of defaultValue
+     */
+    private Number parseJmsProperty(String jmsPropertyKey, Number defaultValue) {
+        String value = jmsConnectionProperties.get(jmsPropertyKey);
+        if (value != null && !value.isEmpty()) {
+            if (defaultValue instanceof Long) {
+                return Long.parseLong(value);
+            } else if (defaultValue instanceof Integer) {
+                return Integer.parseInt(value);
+            }
+        }
+        return defaultValue;
+    }
 
     public synchronized void close() {
 

components/event-publisher/event-output-adapters/org.wso2.carbon.event.output.adapter.jms/src/main/java/org/wso2/carbon/event/output/adapter/jms/internal/util/JMSEventAdapterConstants.java

@@ -44,6 +44,10 @@ public class JMSEventAdapterConstants {
     public static final String ADAPTER_MAX_THREAD_POOL_SIZE_NAME = "maxThread";
     public static final String ADAPTER_KEEP_ALIVE_TIME_NAME = "keepAliveTimeInMillis";
     public static final String ADAPTER_EXECUTOR_JOB_QUEUE_SIZE_NAME = "jobQueueSize";
+    public static final String ADAPTER_MAX_WAIT_NAME = "maxWait";
+    public static final String ADAPTER_MIN_IDLE_NAME = "minIdle";
+    public static final String ADAPTER_MIN_EVICTABLE_IDLE_TIME_NAME = "minEvictableIdleTimeInMillis";
+    public static final String ADAPTER_TIME_BETWEEN_EVICTION_RUNS_NAME = "timeBetweenEvictionRunsInMillis";
     public static final int ADAPTER_MIN_THREAD_POOL_SIZE = 8;
     public static final int ADAPTER_MAX_THREAD_POOL_SIZE = 100;
     public static final int ADAPTER_EXECUTOR_JOB_QUEUE_SIZE = 2000;

components/event-publisher/event-output-adapters/org.wso2.carbon.event.output.adapter.jms/src/test/java/org/wso2/carbon/event/output/adapter/jms/JmsPublisherTestCase.java

@@ -30,6 +30,7 @@ public class JmsPublisherTestCase {
     private OutputEventAdapterConfiguration eventAdapterConfiguration;
     private static final Log LOGGER = LogFactory.getLog(JmsPublisherTestCase.class);
     private static final Path testDir = Paths.get("src", "test", "resources");
+    private Map<String, String> globalProperties;
 
     private void setupCarbonConfig(String tenantName) {
         System.setProperty("carbon.home",
@@ -57,11 +58,15 @@ private JMSEventAdapter getJmsAdaptor() throws NamingException {
         //staticProperties.put("transport.LOGGER.UserName","admin");
         //staticProperties.put("transport.LOGGER.Password","admin");
         eventAdapterConfiguration.setStaticProperties(staticProperties);
-        Map<String, String> globalProperties = new HashMap<>();
+        globalProperties = new HashMap<>();
+        globalProperties.put("minIdle", "0");
+        globalProperties.put("timeBetweenEvictionRunsInMillis", "5000");
+        globalProperties.put("maxWait", "3000");
+        globalProperties.put("maxThread", "100");
         globalProperties.put("keepAliveTimeInMillis", "20000");
-        globalProperties.put("minThread", "8");
+        globalProperties.put("minEvictableIdleTimeInMillis", "3000");
         globalProperties.put("jobQueueSize", "10000");
-        globalProperties.put("maxThread", "100");
+        globalProperties.put("minThread", "8");
 
         return new JMSEventAdapter(eventAdapterConfiguration, globalProperties);
     }
@@ -118,7 +123,7 @@ public void testConnectionFactory() throws Exception {
         adaptorProperties.putAll(eventAdapterConfiguration.getStaticProperties());
         JMSConnectionFactory jmsConnectionFactory = new JMSConnectionFactory(adaptorProperties,
                 eventAdapterConfiguration.getName(), adaptorProperties.get(JMSEventAdapterConstants.ADAPTER_JMS_DESTINATION),
-                1);
+                1, globalProperties);
         jmsConnectionFactory.createConnection();
         jmsConnectionFactory.getConnectionFromPool();
         jmsConnectionFactory.getContext();

features/event-publisher/org.wso2.carbon.event.output.adapter.server.feature/src/main/resources/conf_templates/org.wso2.carbon.event.output.adapter.default.json

@@ -3,6 +3,10 @@
   "output_adapter.jms.max_thread": "100",
   "output_adapter.jms.keep_alive_time": "20000ms",
   "output_adapter.jms.job_queue_size": "10000",
+  "output_adapter.jms.max_wait": "3000",
+  "output_adapter.jms.min_idle": "0",
+  "output_adapter.jms.min_evictable_idle_time": "3000",
+  "output_adapter.jms.time_between_evictions": "5000",
   "output_adapter.email.from_address": "[email protected]",
   "output_adapter.email.username": "abcd",
   "output_adapter.email.password": "xxxx",

features/event-publisher/org.wso2.carbon.event.output.adapter.server.feature/src/main/resources/conf_templates/templates/repository/conf/output-event-adapters.xml.j2

@@ -71,6 +71,10 @@
         <property key="maxThread">{{output_adapter.jms.max_thread}}</property>
         <property key="keepAliveTimeInMillis">{{output_adapter.jms.keep_alive_time}}</property>
         <property key="jobQueueSize">{{output_adapter.jms.job_queue_size}}</property>
+        <property key="maxWait">{{output_adapter.jms.max_wait}}</property>
+        <property key="minIdle">{{output_adapter.jms.min_idle}}</property>
+        <property key="minEvictableIdleTimeInMillis">{{output_adapter.jms.min_evictable_idle_time}}</property>
+        <property key="timeBetweenEvictionRunsMillis">{{output_adapter.jms.time_between_evictions}}</property>
     </adapterConfig>
 
     <adapterConfig type="mqtt">