feat: CE support for connection pooling (#27859)

### Description
> This is a CE support PR for thie connection pooling issue

Ref: https://github.com/appsmithorg/appsmith-ee/pull/2143

Fixes https://github.com/appsmithorg/appsmith-ee/issues/2124

app/server/appsmith-interfaces/src/main/java/com/appsmith/external/connectionpoolconfig/configurations/ConnectionPoolConfig.java

@@ -0,0 +1,3 @@
+package com.appsmith.external.connectionpoolconfig.configurations;
+
+public interface ConnectionPoolConfig extends ConnectionPoolConfigCECompatible {}

app/server/appsmith-interfaces/src/main/java/com/appsmith/external/connectionpoolconfig/configurations/ConnectionPoolConfigCE.java

@@ -0,0 +1,7 @@
+package com.appsmith.external.connectionpoolconfig.configurations;
+
+import reactor.core.publisher.Mono;
+
+public interface ConnectionPoolConfigCE {
+    Mono<Integer> getMaxConnectionPoolSize();
+}

app/server/appsmith-interfaces/src/main/java/com/appsmith/external/connectionpoolconfig/configurations/ConnectionPoolConfigCECompatible.java

@@ -0,0 +1,3 @@
+package com.appsmith.external.connectionpoolconfig.configurations;
+
+public interface ConnectionPoolConfigCECompatible extends ConnectionPoolConfigCE {}

app/server/appsmith-plugins/postgresPlugin/src/main/java/com/external/plugins/PostgresPlugin.java

@@ -1,5 +1,6 @@
 package com.external.plugins;
 
+import com.appsmith.external.connectionpoolconfig.configurations.ConnectionPoolConfig;
 import com.appsmith.external.constants.DataType;
 import com.appsmith.external.datatypes.AppsmithType;
 import com.appsmith.external.dtos.ExecuteActionDTO;
@@ -182,9 +183,11 @@ public class PostgresPlugin extends BasePlugin {
         private static final int PREPARED_STATEMENT_INDEX = 0;
 
         private final SharedConfig sharedConfig;
+        private final ConnectionPoolConfig connectionPoolConfig;
 
-        public PostgresPluginExecutor(SharedConfig sharedConfig) {
+        public PostgresPluginExecutor(SharedConfig sharedConfig, ConnectionPoolConfig connectionPoolConfig) {
             this.sharedConfig = sharedConfig;
+            this.connectionPoolConfig = connectionPoolConfig;
             MAX_SIZE_SUPPORTED = sharedConfig.getMaxResponseSize();
         }
 
@@ -609,9 +612,13 @@ public class PostgresPlugin extends BasePlugin {
                         e.getMessage()));
             }
 
-            return Mono.fromCallable(() -> {
-                        log.debug("Connecting to Postgres db");
-                        return createConnectionPool(datasourceConfiguration);
+            return connectionPoolConfig
+                    .getMaxConnectionPoolSize()
+                    .flatMap(maxPoolSize -> {
+                        return Mono.fromCallable(() -> {
+                            log.debug("Connecting to Postgres db");
+                            return createConnectionPool(datasourceConfiguration, maxPoolSize);
+                        });
                     })
                     .subscribeOn(scheduler);
         }
@@ -1070,7 +1077,8 @@ public class PostgresPlugin extends BasePlugin {
      * @param datasourceConfiguration
      * @return connection pool
      */
-    private static HikariDataSource createConnectionPool(DatasourceConfiguration datasourceConfiguration)
+    private static HikariDataSource createConnectionPool(
+            DatasourceConfiguration datasourceConfiguration, Integer maximumConfigurablePoolSize)
             throws AppsmithPluginException {
         HikariConfig config = new HikariConfig();
 
@@ -1079,7 +1087,13 @@ public class PostgresPlugin extends BasePlugin {
         // Set SSL property
         com.appsmith.external.models.Connection configurationConnection = datasourceConfiguration.getConnection();
         config.setMinimumIdle(MINIMUM_POOL_SIZE);
-        config.setMaximumPoolSize(MAXIMUM_POOL_SIZE);
+
+        int maxPoolSize = MAXIMUM_POOL_SIZE;
+        if (maximumConfigurablePoolSize != null && maximumConfigurablePoolSize >= maxPoolSize) {
+            maxPoolSize = maximumConfigurablePoolSize;
+        }
+
+        config.setMaximumPoolSize(maxPoolSize);
 
         // Set authentication properties
         DBAuth authentication = (DBAuth) datasourceConfiguration.getAuthentication();

app/server/appsmith-plugins/postgresPlugin/src/test/java/com/external/plugins/PostgresPluginTest.java

@@ -1,5 +1,6 @@
 package com.external.plugins;
 
+import com.appsmith.external.connectionpoolconfig.configurations.ConnectionPoolConfig;
 import com.appsmith.external.datatypes.ClientDataType;
 import com.appsmith.external.dtos.ExecuteActionDTO;
 import com.appsmith.external.exceptions.pluginExceptions.AppsmithPluginException;
@@ -77,8 +78,15 @@ public class PostgresPluginTest {
         }
     }
 
+    public class MockConnectionPoolConfig implements ConnectionPoolConfig {
+        @Override
+        public Mono<Integer> getMaxConnectionPoolSize() {
+            return Mono.just(5);
+        }
+    }
+
     PostgresPlugin.PostgresPluginExecutor pluginExecutor =
-            new PostgresPlugin.PostgresPluginExecutor(new MockSharedConfig());
+            new PostgresPlugin.PostgresPluginExecutor(new MockSharedConfig(), new MockConnectionPoolConfig());
 
     @SuppressWarnings("rawtypes") // The type parameter for the container type is just itself and is pseudo-optional.
     @Container

app/server/appsmith-server/src/main/java/com/appsmith/server/connectionpoolconfig/configurations/ConnectionPoolConfigCECompatibleImpl.java

@@ -0,0 +1,8 @@
+package com.appsmith.server.connectionpoolconfig.configurations;
+
+import com.appsmith.external.connectionpoolconfig.configurations.ConnectionPoolConfigCECompatible;
+import org.springframework.stereotype.Component;
+
+@Component
+public class ConnectionPoolConfigCECompatibleImpl extends ConnectionPoolConfigCEImpl
+        implements ConnectionPoolConfigCECompatible {}

app/server/appsmith-server/src/main/java/com/appsmith/server/connectionpoolconfig/configurations/ConnectionPoolConfigCEImpl.java

@@ -0,0 +1,14 @@
+package com.appsmith.server.connectionpoolconfig.configurations;
+
+import com.appsmith.external.connectionpoolconfig.configurations.ConnectionPoolConfigCE;
+import reactor.core.publisher.Mono;
+
+public class ConnectionPoolConfigCEImpl implements ConnectionPoolConfigCE {
+
+    protected static final Integer DEFAULT_MINIMUM_MAX_POOL_SIZE = 5;
+
+    @Override
+    public Mono<Integer> getMaxConnectionPoolSize() {
+        return Mono.just(DEFAULT_MINIMUM_MAX_POOL_SIZE);
+    }
+}

app/server/appsmith-server/src/main/java/com/appsmith/server/connectionpoolconfig/configurations/ConnectionPoolConfigImpl.java

@@ -0,0 +1,9 @@
+package com.appsmith.server.connectionpoolconfig.configurations;
+
+import com.appsmith.external.connectionpoolconfig.configurations.ConnectionPoolConfig;
+import org.springframework.stereotype.Component;
+
+@Component
+public class ConnectionPoolConfigImpl extends ConnectionPoolConfigCECompatibleImpl implements ConnectionPoolConfig {
+    public ConnectionPoolConfigImpl() {}
+}

app/server/appsmith-server/src/test/java/com/appsmith/server/connectionpoolconfig/configurations/ConnectionPoolConfigCETest.java

@@ -0,0 +1,31 @@
+package com.appsmith.server.connectionpoolconfig.configurations;
+
+import com.appsmith.external.connectionpoolconfig.configurations.ConnectionPoolConfig;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.test.context.junit.jupiter.SpringExtension;
+import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+@SpringBootTest
+@ExtendWith(SpringExtension.class)
+public class ConnectionPoolConfigCETest {
+
+    @Autowired
+    ConnectionPoolConfig connectionPoolConfig;
+
+    @Test
+    public void verifyGetMaxConnectionPoolSizeProvidesDefaultValue() {
+        // this is same as default
+        Integer connectionPoolMaxSize = 5;
+
+        Mono<Integer> connectionPoolMaxSizeMono = connectionPoolConfig.getMaxConnectionPoolSize();
+        StepVerifier.create(connectionPoolMaxSizeMono).assertNext(poolSize -> {
+            assertThat(poolSize).isEqualTo(connectionPoolMaxSize);
+        });
+    }
+}