feat: add configurable connection pool support for MySQL plugin (#39148)

## Description
With current implementation, MySQL is configured with HikariCP to have
maximum pool size of 20. This PR makes the connection pool size
configurable from admin settings for MySQL as well, similar to MSSQL
implementation.

This PR adds:
- ConnectionPoolConfig injection in MySqlPluginExecutor
- Support for configurable pool size in datasourceCreate
- Default fallback to 20 connections if not configured
- Test coverage with MockConnectionPoolConfig

Fixes #22525

Link to Devin run:
https://app.devin.ai/sessions/aa7f7e398dae4d24b9bbbac7dc3615b1
Requested by: Sneha

## Automation

/ok-to-test tags="@tag.Datasource"

### 🔍 Cypress test results
> [!CAUTION]  
> If you modify the content in this section, you are likely to disrupt
the CI result for your PR.



<!-- This is an auto-generated comment: Cypress test results  -->
> [!TIP]
> 🟢 🟢 🟢 All cypress tests have passed! 🎉 🎉 🎉
> Workflow run:
<https://github.com/appsmithorg/appsmith/actions/runs/13695631768>
> Commit: 8f5d7e0eecfc3465bc7617546cec5c743fad2414
> <a
href="https://internal.appsmith.com/app/cypress-dashboard/rundetails-65890b3c81d7400d08fa9ee5?branch=master&workflowId=13695631768&attempt=2"
target="_blank">Cypress dashboard</a>.
> Tags: `@tag.Datasource`
> Spec:
> <hr>Thu, 06 Mar 2025 11:41:54 UTC
<!-- end of auto-generated comment: Cypress test results  -->

---------

Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Co-authored-by: Sneha Bodke <sneha@appsmith.com>
Co-authored-by: “sneha122” <“sneha@appsmith.com”>

app/server/appsmith-plugins/mysqlPlugin/src/main/java/com/external/plugins/MySqlPlugin.java

@@ -1,5 +1,6 @@
 package com.external.plugins;
 
+import com.appsmith.external.configurations.connectionpool.ConnectionPoolConfig;
 import com.appsmith.external.datatypes.AppsmithType;
 import com.appsmith.external.dtos.ExecuteActionDTO;
 import com.appsmith.external.exceptions.pluginExceptions.AppsmithPluginError;
@@ -161,6 +162,11 @@ public class MySqlPlugin extends BasePlugin {
 
         private static final int PREPARED_STATEMENT_INDEX = 0;
         private final Scheduler scheduler = Schedulers.boundedElastic();
+        private final ConnectionPoolConfig connectionPoolConfig;
+
+        public MySqlPluginExecutor(ConnectionPoolConfig connectionPoolConfig) {
+            this.connectionPoolConfig = connectionPoolConfig;
+        }
 
         /**
          * Instead of using the default executeParametrized provided by pluginExecutor, this implementation affords an opportunity
@@ -668,9 +674,12 @@ public class MySqlPlugin extends BasePlugin {
                 try {
                     connectionContext = getConnectionContext(
                             datasourceConfiguration, CONNECTION_METHOD_INDEX, MYSQL_DEFAULT_PORT, ConnectionPool.class);
-                    ConnectionPool pool = getNewConnectionPool(datasourceConfiguration, connectionContext);
-                    connectionContext.setConnection(pool);
-                    return Mono.just(connectionContext);
+                    return connectionPoolConfig.getMaxConnectionPoolSize().flatMap(maxPoolSize -> {
+                        ConnectionPool pool =
+                                getNewConnectionPool(datasourceConfiguration, connectionContext, maxPoolSize);
+                        connectionContext.setConnection(pool);
+                        return Mono.just(connectionContext);
+                    });
                 } catch (AppsmithPluginException e) {
                     return Mono.error(e);
                 }

app/server/appsmith-plugins/mysqlPlugin/src/main/java/com/external/utils/MySqlDatasourceUtils.java

@@ -245,7 +245,7 @@ public class MySqlDatasourceUtils {
     }
 
     public static ConnectionPool getNewConnectionPool(
-            DatasourceConfiguration datasourceConfiguration, ConnectionContext connectionContext)
+            DatasourceConfiguration datasourceConfiguration, ConnectionContext connectionContext, Integer maxPoolSize)
             throws AppsmithPluginException {
         ConnectionFactoryOptions.Builder ob = getBuilder(datasourceConfiguration, connectionContext);
         ob = addSslOptionsToBuilder(datasourceConfiguration, ob);
@@ -260,7 +260,7 @@ public class MySqlDatasourceUtils {
          */
         ConnectionPoolConfiguration configuration = ConnectionPoolConfiguration.builder(connectionFactory)
                 .maxIdleTime(MAX_IDLE_TIME)
-                .maxSize(MAX_CONNECTION_POOL_SIZE)
+                .maxSize(maxPoolSize != null ? maxPoolSize : MAX_CONNECTION_POOL_SIZE)
                 .backgroundEvictionInterval(BACKGROUND_EVICTION_TIME)
                 .maxLifeTime(MAX_LIFE_TIME)
                 .build();

app/server/appsmith-plugins/mysqlPlugin/src/test/java/com/external/plugins/MySQLDatasourceValidationTest.java

@@ -1,5 +1,6 @@
 package com.external.plugins;
 
+import com.appsmith.external.configurations.connectionpool.ConnectionPoolConfig;
 import com.appsmith.external.models.Connection;
 import com.appsmith.external.models.DBAuth;
 import com.appsmith.external.models.DatasourceConfiguration;
@@ -25,7 +26,15 @@ import static com.appsmith.external.models.Connection.Mode.READ_WRITE;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 public class MySQLDatasourceValidationTest {
-    static MySqlPlugin.MySqlPluginExecutor pluginExecutor = new MySqlPlugin.MySqlPluginExecutor();
+    private static class MockConnectionPoolConfig implements ConnectionPoolConfig {
+        @Override
+        public Mono<Integer> getMaxConnectionPoolSize() {
+            return Mono.just(5);
+        }
+    }
+
+    static MySqlPlugin.MySqlPluginExecutor pluginExecutor =
+            new MySqlPlugin.MySqlPluginExecutor(new MockConnectionPoolConfig());
 
     private DatasourceConfiguration getDatasourceConfigurationWithStandardConnectionMethod() {
         DatasourceConfiguration datasourceConfiguration = new DatasourceConfiguration();

app/server/appsmith-plugins/mysqlPlugin/src/test/java/com/external/plugins/MySqlPluginTest.java

@@ -1,5 +1,6 @@
 package com.external.plugins;
 
+import com.appsmith.external.configurations.connectionpool.ConnectionPoolConfig;
 import com.appsmith.external.datatypes.ClientDataType;
 import com.appsmith.external.dtos.ExecuteActionDTO;
 import com.appsmith.external.exceptions.pluginExceptions.AppsmithPluginException;
@@ -76,7 +77,15 @@ import static reactor.core.publisher.Mono.zip;
 @Testcontainers
 public class MySqlPluginTest {
 
-    static MySqlPlugin.MySqlPluginExecutor pluginExecutor = new MySqlPlugin.MySqlPluginExecutor();
+    private static class MockConnectionPoolConfig implements ConnectionPoolConfig {
+        @Override
+        public Mono<Integer> getMaxConnectionPoolSize() {
+            return Mono.just(5);
+        }
+    }
+
+    static MySqlPlugin.MySqlPluginExecutor pluginExecutor =
+            new MySqlPlugin.MySqlPluginExecutor(new MockConnectionPoolConfig());
 
     ConnectionContext<ConnectionPool> instanceConnectionContext;
 
@@ -1318,7 +1327,7 @@ public class MySqlPluginTest {
 
     @Test
     public void testNullObjectWithPreparedStatement() {
-        pluginExecutor = spy(new MySqlPlugin.MySqlPluginExecutor());
+        pluginExecutor = spy(new MySqlPlugin.MySqlPluginExecutor(new MockConnectionPoolConfig()));
         doReturn(false).when(pluginExecutor).isIsOperatorUsed(any());
         DatasourceConfiguration dsConfig = createDatasourceConfiguration();
         Mono<ConnectionContext<ConnectionPool>> connectionContextMono = pluginExecutor

app/server/appsmith-plugins/mysqlPlugin/src/test/java/com/external/plugins/MySqlStaleConnectionErrorMessageTest.java

@@ -1,5 +1,6 @@
 package com.external.plugins;
 
+import com.appsmith.external.configurations.connectionpool.ConnectionPoolConfig;
 import com.appsmith.external.dtos.ExecuteActionDTO;
 import com.appsmith.external.exceptions.pluginExceptions.StaleConnectionException;
 import com.appsmith.external.models.ActionConfiguration;
@@ -26,7 +27,15 @@ import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
 public class MySqlStaleConnectionErrorMessageTest {
-    static MySqlPlugin.MySqlPluginExecutor pluginExecutor = new MySqlPlugin.MySqlPluginExecutor();
+    private static class MockConnectionPoolConfig implements ConnectionPoolConfig {
+        @Override
+        public Mono<Integer> getMaxConnectionPoolSize() {
+            return Mono.just(5);
+        }
+    }
+
+    static MySqlPlugin.MySqlPluginExecutor pluginExecutor =
+            new MySqlPlugin.MySqlPluginExecutor(new MockConnectionPoolConfig());
     static MySqlDatasourceUtils mysqlDatasourceUtils = new MySqlDatasourceUtils();
 
     @Test