Note in the context of our Beam pipelines, we only serialize `DataSourceConfig` objects and
+ * create `DataSource` from them when necessary; i.e., do not serialize the returned `DataSource`
+ * in our code. Some Beam libraries like JdbcIO may serialize `DataSource` but they should
+ * properly handle the singleton pattern needed for pooled data-sources.
+ *
+ * @param config the JDBC connection information
+ * @param initialPoolSize initial pool size if a new pool is created.
+ * @param jdbcMaxPoolSize maximum pool size if a new pool is created.
+ * @return a pooling DataSource for the given DB config
+ */
+ public DataSource getPooledDataSource(
+ DataSourceConfig config, int initialPoolSize, int jdbcMaxPoolSize)
+ throws PropertyVetoException {
+ dataSources.computeIfAbsent(config, c -> createNewPool(c, initialPoolSize, jdbcMaxPoolSize));
+ return dataSources.get(config);
+ }
+
+ private static DataSource createNewPool(
+ DataSourceConfig config, int initialPoolSize, int jdbcMaxPoolSize) {
+ log.info(
+ "Creating a JDBC connection pool for "
+ + config.jdbcUrl()
+ + " with driver class "
+ + config.jdbcDriverClass()
+ + " and max pool size "
+ + jdbcMaxPoolSize);
+ Preconditions.checkArgument(
+ initialPoolSize <= jdbcMaxPoolSize,
+ "initialPoolSize cannot be larger than jdbcMaxPoolSize");
+ // Note caching of these connection-pools is important beyond just performance benefits. If a
+ // `ComboPooledDataSource` goes out of scope without calling `close()` on it, then it can leak
+ // connections (and memory) as its threads are not killed and can hold those objects; this was
+ // the case even with `setNumHelperThreads(0)`.
+ ComboPooledDataSource comboPooledDataSource = new ComboPooledDataSource();
+ try {
+ comboPooledDataSource.setDriverClass(config.jdbcDriverClass());
+ } catch (PropertyVetoException e) {
+ String errorMes = "Error in setting the JDBC driver class " + config.jdbcDriverClass();
+ log.error(errorMes);
+ throw new IllegalArgumentException(errorMes, e);
+ }
+ comboPooledDataSource.setJdbcUrl(config.jdbcUrl());
+ comboPooledDataSource.setUser(config.dbUser());
+ comboPooledDataSource.setPassword(config.dbPassword());
+ comboPooledDataSource.setMaxPoolSize(jdbcMaxPoolSize);
+ comboPooledDataSource.setInitialPoolSize(initialPoolSize);
+ // Setting an idle time to reduce the number of connections when idle.
+ comboPooledDataSource.setMaxIdleTime(30);
+ // Lowering the minimum pool size to limit the number of connections if multiple pools are
+ // created for the same DB.
+ comboPooledDataSource.setMinPoolSize(1);
+ return comboPooledDataSource;
+ }
+
+ // TODO we should `close()` connection pools on application shutdown.
+
+ public static DataSourceConfig dbConfigToDataSourceConfig(DatabaseConfiguration config) {
+ return DataSourceConfig.create(
+ config.getJdbcDriverClass(),
+ config.makeJdbsUrlFromConfig(),
+ config.getDatabaseUser(),
+ config.getDatabasePassword());
+ }
+
+ /**
+ * This is to identify a database connection pool. We use instances of these to cache connection
+ * pools and to impose a Singleton pattern per connection config (hence AutoValue).
+ */
+ @AutoValue
+ public abstract static class DataSourceConfig {
+ abstract String jdbcDriverClass();
+
+ abstract String jdbcUrl();
+
+ abstract String dbUser();
+
+ abstract String dbPassword();
+
+ static DataSourceConfig create(
+ String jdbcDriverClass, String jdbcUrl, String dbUser, String dbPassword) {
+ return new AutoValue_JdbcConnectionPools_DataSourceConfig(
+ jdbcDriverClass, jdbcUrl, dbUser, dbPassword);
+ }
+ }
+}
diff --git a/pipelines/common/src/main/java/org/openmrs/analytics/JdbcConnectionUtil.java b/pipelines/common/src/main/java/org/openmrs/analytics/JdbcConnectionUtil.java
deleted file mode 100644
index ed1cdf109..000000000
--- a/pipelines/common/src/main/java/org/openmrs/analytics/JdbcConnectionUtil.java
+++ /dev/null
@@ -1,77 +0,0 @@
-/*
- * Copyright 2020-2023 Google LLC
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.openmrs.analytics;
-
-import com.google.common.base.Preconditions;
-import com.mchange.v2.c3p0.ComboPooledDataSource;
-import java.beans.PropertyVetoException;
-import java.sql.Connection;
-import java.sql.SQLException;
-import java.sql.Statement;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class JdbcConnectionUtil {
- private static final Logger log = LoggerFactory.getLogger(JdbcConnectionUtil.class);
-
- private final ComboPooledDataSource comboPooledDataSource;
-
- JdbcConnectionUtil(
- String jdbcDriverClass,
- String jdbcUrl,
- String dbUser,
- String dbPassword,
- int initialPoolSize,
- int jdbcMaxPoolSize)
- throws PropertyVetoException {
- log.info(
- "Creating a JdbcConnectionUtil for "
- + jdbcUrl
- + " with driver class "
- + jdbcDriverClass
- + " and max pool size "
- + jdbcMaxPoolSize);
- Preconditions.checkArgument(
- initialPoolSize <= jdbcMaxPoolSize,
- "initialPoolSize cannot be larger than jdbcMaxPoolSize");
- comboPooledDataSource = new ComboPooledDataSource();
- comboPooledDataSource.setDriverClass(jdbcDriverClass);
- comboPooledDataSource.setJdbcUrl(jdbcUrl);
- comboPooledDataSource.setUser(dbUser);
- comboPooledDataSource.setPassword(dbPassword);
- comboPooledDataSource.setMaxPoolSize(jdbcMaxPoolSize);
- comboPooledDataSource.setInitialPoolSize(initialPoolSize);
- // Setting an idle time to reduce the number of connections when idle.
- comboPooledDataSource.setMaxIdleTime(30);
- }
-
- public Statement createStatement() throws SQLException {
- Connection con = getDataSource().getConnection();
- return con.createStatement();
- }
-
- public void closeConnection(Statement stmt) throws SQLException {
- if (stmt != null) {
- Connection con = stmt.getConnection();
- stmt.close();
- con.close();
- }
- }
-
- public ComboPooledDataSource getDataSource() {
- return comboPooledDataSource;
- }
-}
diff --git a/pipelines/common/src/main/java/org/openmrs/analytics/model/DatabaseConfiguration.java b/pipelines/common/src/main/java/org/openmrs/analytics/model/DatabaseConfiguration.java
index faca8e50d..2e30c17a3 100644
--- a/pipelines/common/src/main/java/org/openmrs/analytics/model/DatabaseConfiguration.java
+++ b/pipelines/common/src/main/java/org/openmrs/analytics/model/DatabaseConfiguration.java
@@ -32,6 +32,7 @@
public class DatabaseConfiguration {
// General configuration parameters that needs to be exposed beyond Debezium.
+ private String jdbcDriverClass;
private String databaseService;
private String databaseHostName;
private String databasePort;
diff --git a/pipelines/common/src/test/java/org/openmrs/analytics/JdbcConnectionUtilTest.java b/pipelines/common/src/test/java/org/openmrs/analytics/JdbcConnectionUtilTest.java
deleted file mode 100644
index 9b25b3637..000000000
--- a/pipelines/common/src/test/java/org/openmrs/analytics/JdbcConnectionUtilTest.java
+++ /dev/null
@@ -1,74 +0,0 @@
-/*
- * Copyright 2020-2022 Google LLC
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.openmrs.analytics;
-
-import static org.junit.Assert.assertTrue;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
-
-import java.beans.PropertyVetoException;
-import java.sql.Connection;
-import java.sql.SQLException;
-import java.sql.Statement;
-import org.junit.Assert;
-import org.junit.Before;
-import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.mockito.junit.MockitoJUnitRunner;
-
-@RunWith(MockitoJUnitRunner.class)
-public class JdbcConnectionUtilTest {
-
- private Connection mockConnection;
-
- private Statement mockStatement;
-
- private JdbcConnectionUtil jdbcConnectionUtil;
-
- @Before
- public void setup() throws PropertyVetoException, SQLException {
- mockStatement = mock(Statement.class);
- mockConnection = mock(Connection.class);
- when(mockStatement.getConnection()).thenReturn(mockConnection);
- jdbcConnectionUtil = new JdbcConnectionUtil("random", "random", "omar", "123", 3, 60);
- }
-
- @Test
- public void testSetIncorrectJdbcPoolSize() throws IllegalArgumentException {
- IllegalArgumentException thrown =
- Assert.assertThrows(
- IllegalArgumentException.class,
- () -> new JdbcConnectionUtil("random", "random", "omar", "123", 4, 2));
-
- assertTrue(
- thrown.getMessage().contains("initialPoolSize cannot be larger than jdbcMaxPoolSize"));
- }
-
- @Test
- public void testCloseConnection() throws PropertyVetoException, SQLException {
- jdbcConnectionUtil.closeConnection(mockStatement);
- verify(mockStatement, times(1)).close();
- }
-
- @Test
- public void testCloseConnectionNullStatement() throws PropertyVetoException, SQLException {
- jdbcConnectionUtil.closeConnection(null);
- verify(mockStatement, times(0)).close();
- verify(mockConnection, times(0)).close();
- }
-}
diff --git a/pipelines/controller/config/hapi-postgres-config.json b/pipelines/controller/config/hapi-postgres-config.json
index 489d2ba77..0f3224edb 100644
--- a/pipelines/controller/config/hapi-postgres-config.json
+++ b/pipelines/controller/config/hapi-postgres-config.json
@@ -1,4 +1,5 @@
{
+ "jdbcDriverClass": "org.postgresql.Driver",
"databaseService" : "postgresql",
"databaseHostName" : "172.17.0.1",
"databasePort" : "5432",
diff --git a/pipelines/pom.xml b/pipelines/pom.xml
index 126865084..53a0c05bb 100644
--- a/pipelines/pom.xml
+++ b/pipelines/pom.xml
@@ -47,6 +47,7 @@