[GOBBLIN-1753] Migrate DB connection pool from o.a.commons.dbcp/dbcp2…

… to HikariCP (#3613)

* [GOBBLIN-1753] Migrate DB connection pool from o.a.commons.dbcp/dbcp2 to HikariCP

* minor change to use `Duration`

gobblin-api/src/main/java/org/apache/gobblin/configuration/ConfigurationKeys.java

@@ -75,10 +75,6 @@ public class ConfigurationKeys {
   // DB state store configuration
   public static final String STATE_STORE_DB_JDBC_DRIVER_KEY = "state.store.db.jdbc.driver";
   public static final String DEFAULT_STATE_STORE_DB_JDBC_DRIVER = "com.mysql.jdbc.Driver";
-  // min idle time for eviction
-  public static final String STATE_STORE_DB_CONN_MIN_EVICTABLE_IDLE_TIME_KEY =
-      "state.store.db.conn.min.evictable.idle.time";
-  public static final long DEFAULT_STATE_STORE_DB_CONN_MIN_EVICTABLE_IDLE_TIME = 5 * 60 * 1000;
   public static final String STATE_STORE_DB_URL_KEY = "state.store.db.url";
   public static final String STATE_STORE_DB_USER_KEY = "state.store.db.user";
   public static final String STATE_STORE_DB_PASSWORD_KEY = "state.store.db.password";

gobblin-core/build.gradle

@@ -32,7 +32,6 @@ dependencies {
 
   compile externalDependency.avroMapredH2
   compile externalDependency.commonsCodec
-  compile externalDependency.commonsDbcp
   compile externalDependency.commonsMath
   compile externalDependency.commonsHttpClient
   compile externalDependency.avro
@@ -46,6 +45,7 @@ dependencies {
   compile externalDependency.jsch
   compile externalDependency.commonsLang3
   compile externalDependency.commonsIo
+  compile externalDependency.hikariCP
   compile externalDependency.hiveExec
   compile externalDependency.hiveSerDe
   compile externalDependency.httpclient

gobblin-data-management/src/test/java/org/apache/gobblin/data/management/retention/CleanableMysqlDatasetStoreDatasetTest.java

@@ -20,7 +20,6 @@
 import java.io.IOException;
 import java.util.List;
 
-import org.apache.commons.dbcp.BasicDataSource;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
 import org.testng.Assert;
@@ -29,6 +28,7 @@
 
 import com.typesafe.config.Config;
 import com.typesafe.config.ConfigValueFactory;
+import com.zaxxer.hikari.HikariDataSource;
 
 import org.apache.gobblin.config.ConfigBuilder;
 import org.apache.gobblin.configuration.ConfigurationKeys;
@@ -78,15 +78,15 @@ public void setUp() throws Exception {
     this.testMetastoreDatabase = TestMetastoreDatabaseFactory.get();
     String jdbcUrl = this.testMetastoreDatabase.getJdbcUrl();
     ConfigBuilder configBuilder = ConfigBuilder.create();
-    BasicDataSource mySqlDs = new BasicDataSource();
+    HikariDataSource dataSource = new HikariDataSource();
 
-    mySqlDs.setDriverClassName(ConfigurationKeys.DEFAULT_STATE_STORE_DB_JDBC_DRIVER);
-    mySqlDs.setDefaultAutoCommit(false);
-    mySqlDs.setUrl(jdbcUrl);
-    mySqlDs.setUsername(TEST_USER);
-    mySqlDs.setPassword(TEST_PASSWORD);
+    dataSource.setDriverClassName(ConfigurationKeys.DEFAULT_STATE_STORE_DB_JDBC_DRIVER);
+    dataSource.setAutoCommit(false);
+    dataSource.setJdbcUrl(jdbcUrl);
+    dataSource.setUsername(TEST_USER);
+    dataSource.setPassword(TEST_PASSWORD);
 
-    this.dbJobStateStore = new MysqlStateStore<>(mySqlDs, TEST_STATE_STORE, false, JobState.class);
+    this.dbJobStateStore = new MysqlStateStore<>(dataSource, TEST_STATE_STORE, false, JobState.class);
 
     configBuilder.addPrimitive("selection.timeBased.lookbackTime", "10m");
     configBuilder.addPrimitive(ConfigurationKeys.STATE_STORE_TYPE_KEY, "mysql");

gobblin-data-management/src/test/java/org/apache/gobblin/data/management/retention/sql/SqlBasedRetentionPoc.java

@@ -22,7 +22,8 @@
 import java.sql.SQLException;
 import java.sql.Timestamp;
 
-import org.apache.commons.dbcp.BasicDataSource;
+import com.zaxxer.hikari.HikariDataSource;
+
 import org.apache.commons.lang.StringUtils;
 import org.apache.hadoop.fs.Path;
 import org.joda.time.DateTime;
@@ -42,7 +43,7 @@ public class SqlBasedRetentionPoc {
   private static final int TWO_YEARS_IN_DAYS = 365 * 2;
   private static final String DAILY_PARTITION_PATTERN = "yyyy/MM/dd";
 
-  private BasicDataSource basicDataSource;
+  private HikariDataSource dataSource;
   private Connection connection;
 
   /**
@@ -55,11 +56,11 @@ public class SqlBasedRetentionPoc {
    */
   @BeforeClass
   public void setup() throws SQLException {
-    basicDataSource = new BasicDataSource();
-    basicDataSource.setDriverClassName("org.apache.derby.jdbc.EmbeddedDriver");
-    basicDataSource.setUrl("jdbc:derby:memory:derbypoc;create=true");
+    dataSource = new HikariDataSource();
+    dataSource.setDriverClassName("org.apache.derby.jdbc.EmbeddedDriver");
+    dataSource.setJdbcUrl("jdbc:derby:memory:derbypoc;create=true");
 
-    Connection connection = basicDataSource.getConnection();
+    Connection connection = dataSource.getConnection();
     connection.setAutoCommit(false);
     this.connection = connection;
 
@@ -72,7 +73,7 @@ public void setup() throws SQLException {
 
   @AfterClass
   public void cleanUp() throws Exception {
-    basicDataSource.close();
+    dataSource.close();
   }
 
   /**

gobblin-metastore/build.gradle

@@ -25,12 +25,12 @@ dependencies {
     compile externalDependency.guava
     compile externalDependency.slf4j
     compile externalDependency.pegasus.data
-    compile externalDependency.commonsDbcp
     compile externalDependency.commonsLang
     compile externalDependency.commonsLang3
     compile externalDependency.guice
     compile externalDependency.javaxInject
     compile externalDependency.jodaTime
+    compile externalDependency.hikariCP
     compile externalDependency.httpclient
     compile externalDependency.flyway
     compile externalDependency.commonsConfiguration

gobblin-metastore/src/main/java/org/apache/gobblin/metastore/JobHistoryDataSourceProvider.java

@@ -39,24 +39,32 @@ public class JobHistoryDataSourceProvider extends org.apache.gobblin.util.jdbc.D
 
   @Inject
   public JobHistoryDataSourceProvider(@Named("dataSourceProperties") Properties properties) {
-    this.basicDataSource.setDriverClassName(properties.getProperty(ConfigurationKeys.JOB_HISTORY_STORE_JDBC_DRIVER_KEY,
+    this.dataSource.setDriverClassName(properties.getProperty(ConfigurationKeys.JOB_HISTORY_STORE_JDBC_DRIVER_KEY,
         ConfigurationKeys.DEFAULT_JOB_HISTORY_STORE_JDBC_DRIVER));
 
     // Set validation query to verify connection
     if (!Boolean.parseBoolean(properties.getProperty(SKIP_VALIDATION_QUERY, "false"))) {
       // MySQL server can timeout a connection so need to validate connections before use
       final String validationQuery = MysqlDataSourceUtils.QUERY_CONNECTION_IS_VALID_AND_NOT_READONLY;
       LOG.info("setting `DataSource` validation query: '" + validationQuery + "'");
-      this.basicDataSource.setValidationQuery(validationQuery);
-      this.basicDataSource.setTestOnBorrow(true);
-      this.basicDataSource.setTimeBetweenEvictionRunsMillis(Duration.ofSeconds(60).toMillis());
+      // TODO: revisit following verification of successful connection pool migration:
+      //   If your driver supports JDBC4 we strongly recommend not setting this property. This is for "legacy" drivers
+      //   that do not support the JDBC4 Connection.isValid() API; see:
+      //   https://github.com/brettwooldridge/HikariCP#gear-configuration-knobs-baby
+      this.dataSource.setConnectionTestQuery(validationQuery);
+      this.dataSource.setIdleTimeout(Duration.ofSeconds(60).toMillis());
     }
 
-    this.basicDataSource.setUrl(properties.getProperty(ConfigurationKeys.JOB_HISTORY_STORE_URL_KEY));
+    this.dataSource.setJdbcUrl(properties.getProperty(ConfigurationKeys.JOB_HISTORY_STORE_URL_KEY));
+    // TODO: revisit following verification of successful connection pool migration:
+    //   whereas `o.a.commons.dbcp.BasicDataSource` defaults min idle conns to 0, hikari defaults to 10.
+    //   perhaps non-zero would have desirable runtime perf, but anything >0 currently fails unit tests (even 1!);
+    //   (so experimenting with a higher number would first require adjusting tests)
+    this.dataSource.setMinimumIdle(0);
     if (properties.containsKey(ConfigurationKeys.JOB_HISTORY_STORE_USER_KEY)
         && properties.containsKey(ConfigurationKeys.JOB_HISTORY_STORE_PASSWORD_KEY)) {
-      this.basicDataSource.setUsername(properties.getProperty(ConfigurationKeys.JOB_HISTORY_STORE_USER_KEY));
-      this.basicDataSource.setPassword(PasswordManager.getInstance(properties)
+      this.dataSource.setUsername(properties.getProperty(ConfigurationKeys.JOB_HISTORY_STORE_USER_KEY));
+      this.dataSource.setPassword(PasswordManager.getInstance(properties)
           .readPassword(properties.getProperty(ConfigurationKeys.JOB_HISTORY_STORE_PASSWORD_KEY)));
     }
   }

gobblin-metastore/src/main/java/org/apache/gobblin/metastore/MysqlDagStateStoreFactory.java

@@ -17,7 +17,7 @@
 
 package org.apache.gobblin.metastore;
 
-import org.apache.commons.dbcp.BasicDataSource;
+import javax.sql.DataSource;
 
 import com.typesafe.config.Config;
 
@@ -37,10 +37,10 @@ public <T extends State> MysqlDagStore<T> createStateStore(Config config, Class<
         ConfigurationKeys.DEFAULT_STATE_STORE_COMPRESSED_VALUES);
 
     try {
-      BasicDataSource basicDataSource = MysqlDataSourceFactory.get(config,
+      DataSource dataSource = MysqlDataSourceFactory.get(config,
           SharedResourcesBrokerFactory.getImplicitBroker());
 
-      return new MysqlDagStore<>(basicDataSource, stateStoreTableName, compressedValues, stateClass);
+      return new MysqlDagStore<>(dataSource, stateStoreTableName, compressedValues, stateClass);
     } catch (Exception e) {
       throw new RuntimeException("Failed to create MysqlDagStore with factory", e);
     }

gobblin-metastore/src/main/java/org/apache/gobblin/metastore/MysqlDataSourceFactory.java

@@ -19,7 +19,7 @@
 
 import java.io.IOException;
 
-import org.apache.commons.dbcp.BasicDataSource;
+import javax.sql.DataSource;
 
 import com.typesafe.config.Config;
 
@@ -35,24 +35,25 @@
 import lombok.extern.slf4j.Slf4j;
 
 /**
- * A {@link SharedResourceFactory} for creating {@link BasicDataSource}s.
+ * A {@link SharedResourceFactory} for creating {@link DataSource}s.
  *
- * The factory creates a {@link BasicDataSource} with the config.
+ * The factory creates a {@link DataSource} with the config.
  */
 @Slf4j
 public class MysqlDataSourceFactory<S extends ScopeType<S>>
-    implements SharedResourceFactory<BasicDataSource, MysqlDataSourceKey, S> {
+    implements SharedResourceFactory<DataSource, MysqlDataSourceKey, S> {
 
+  // WARNING: now a misnomer, but retained for legacy compatibility, despite move from `o.a.commons.dbcp.BasicDataSource` to `HikariCP`
   public static final String FACTORY_NAME = "basicDataSource";
 
   /**
-   * Get a {@link BasicDataSource} based on the config
+   * Get a {@link DataSource} based on the config
    * @param config configuration
    * @param broker broker
-   * @return a {@link BasicDataSource}
+   * @return a {@link DataSource}
    * @throws IOException
    */
-  public static <S extends ScopeType<S>> BasicDataSource get(Config config,
+  public static <S extends ScopeType<S>> DataSource get(Config config,
       SharedResourcesBroker<S> broker) throws IOException {
     try {
       return broker.getSharedResource(new MysqlDataSourceFactory<S>(),
@@ -68,12 +69,12 @@ public String getName() {
   }
 
   @Override
-  public SharedResourceFactoryResponse<BasicDataSource> createResource(SharedResourcesBroker<S> broker,
+  public SharedResourceFactoryResponse<DataSource> createResource(SharedResourcesBroker<S> broker,
     ScopedConfigView<S, MysqlDataSourceKey> config) throws NotConfiguredException {
     MysqlDataSourceKey key = config.getKey();
     Config configuration = key.getConfig();
 
-    BasicDataSource dataSource = MysqlStateStore.newDataSource(configuration);
+    DataSource dataSource = MysqlStateStore.newDataSource(configuration);
 
     return new ResourceInstance<>(dataSource);
   }

gobblin-metastore/src/main/java/org/apache/gobblin/metastore/MysqlDataSourceKey.java

@@ -17,7 +17,7 @@
 
 package org.apache.gobblin.metastore;
 
-import org.apache.commons.dbcp.BasicDataSource;
+import javax.sql.DataSource;
 
 import com.typesafe.config.Config;
 
@@ -26,7 +26,7 @@
 import lombok.Getter;
 
 /**
- * {@link SharedResourceKey} for requesting {@link BasicDataSource}s from a
+ * {@link SharedResourceKey} for requesting {@link DataSource}s from a
  * {@link org.apache.gobblin.broker.iface.SharedResourceFactory}
  */
 @Getter

gobblin-metastore/src/main/java/org/apache/gobblin/metastore/MysqlJobStatusStateStoreFactory.java

@@ -17,7 +17,7 @@
 
 package org.apache.gobblin.metastore;
 
-import org.apache.commons.dbcp.BasicDataSource;
+import javax.sql.DataSource;
 
 import com.typesafe.config.Config;
 
@@ -36,10 +36,10 @@ public <T extends State> MysqlJobStatusStateStore<T> createStateStore(Config con
         ConfigurationKeys.DEFAULT_STATE_STORE_COMPRESSED_VALUES);
 
     try {
-      BasicDataSource basicDataSource = MysqlDataSourceFactory.get(config,
+      DataSource dataSource = MysqlDataSourceFactory.get(config,
           SharedResourcesBrokerFactory.getImplicitBroker());
 
-      return new MysqlJobStatusStateStore<>(basicDataSource, stateStoreTableName, compressedValues, stateClass);
+      return new MysqlJobStatusStateStore<>(dataSource, stateStoreTableName, compressedValues, stateClass);
     } catch (Exception e) {
       throw new RuntimeException("Failed to create MysqlStateStore with factory", e);
     }

gobblin-metastore/src/main/java/org/apache/gobblin/metastore/MysqlStateStore.java

@@ -38,20 +38,19 @@
 import java.util.List;
 import java.util.zip.GZIPInputStream;
 import java.util.zip.GZIPOutputStream;
+import javax.sql.DataSource;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import org.apache.commons.dbcp.BasicDataSource;
 import org.apache.hadoop.io.Text;
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Predicate;
 import com.google.common.base.Strings;
 import com.google.common.collect.Lists;
 import com.typesafe.config.Config;
-
-import javax.sql.DataSource;
+import com.zaxxer.hikari.HikariDataSource;
 
 import org.apache.gobblin.configuration.ConfigurationKeys;
 import org.apache.gobblin.configuration.State;
@@ -194,33 +193,38 @@ protected String getCreateJobStateTableTemplate() {
   }
 
   /**
-   * creates a new {@link BasicDataSource}
+   * creates a new {@link DataSource}
    * @param config the properties used for datasource instantiation
    * @return
    */
-  public static BasicDataSource newDataSource(Config config) {
-    BasicDataSource basicDataSource = new BasicDataSource();
+  public static DataSource newDataSource(Config config) {
+    HikariDataSource dataSource = new HikariDataSource();
     PasswordManager passwordManager = PasswordManager.getInstance(ConfigUtils.configToProperties(config));
 
-    basicDataSource.setDriverClassName(ConfigUtils.getString(config, ConfigurationKeys.STATE_STORE_DB_JDBC_DRIVER_KEY,
+    dataSource.setDriverClassName(ConfigUtils.getString(config, ConfigurationKeys.STATE_STORE_DB_JDBC_DRIVER_KEY,
         ConfigurationKeys.DEFAULT_STATE_STORE_DB_JDBC_DRIVER));
     // MySQL server can timeout a connection so need to validate connections before use
     final String validationQuery = MysqlDataSourceUtils.QUERY_CONNECTION_IS_VALID_AND_NOT_READONLY;
     LOG.info("setting `DataSource` validation query: '" + validationQuery + "'");
-    basicDataSource.setValidationQuery(validationQuery);
-    basicDataSource.setTestOnBorrow(true);
-    basicDataSource.setDefaultAutoCommit(false);
-    basicDataSource.setTimeBetweenEvictionRunsMillis(Duration.ofSeconds(60).toMillis());
-    basicDataSource.setUrl(config.getString(ConfigurationKeys.STATE_STORE_DB_URL_KEY));
-    basicDataSource.setUsername(passwordManager.readPassword(
+    // TODO: revisit following verification of successful connection pool migration:
+    //   If your driver supports JDBC4 we strongly recommend not setting this property. This is for "legacy" drivers
+    //   that do not support the JDBC4 Connection.isValid() API; see:
+    //   https://github.com/brettwooldridge/HikariCP#gear-configuration-knobs-baby
+    dataSource.setConnectionTestQuery(validationQuery);
+    dataSource.setAutoCommit(false);
+    dataSource.setIdleTimeout(Duration.ofSeconds(60).toMillis());
+    dataSource.setJdbcUrl(config.getString(ConfigurationKeys.STATE_STORE_DB_URL_KEY));
+    // TODO: revisit following verification of successful connection pool migration:
+    //   whereas `o.a.commons.dbcp.BasicDataSource` defaults min idle conns to 0, hikari defaults to 10.
+    //   perhaps non-zero would have desirable runtime perf, but anything >0 currently fails unit tests (even 1!);
+    //   (so experimenting with a higher number would first require adjusting tests)
+    dataSource.setMinimumIdle(0);
+    dataSource.setUsername(passwordManager.readPassword(
         config.getString(ConfigurationKeys.STATE_STORE_DB_USER_KEY)));
-    basicDataSource.setPassword(passwordManager.readPassword(
+    dataSource.setPassword(passwordManager.readPassword(
         config.getString(ConfigurationKeys.STATE_STORE_DB_PASSWORD_KEY)));
-    basicDataSource.setMinEvictableIdleTimeMillis(
-        ConfigUtils.getLong(config, ConfigurationKeys.STATE_STORE_DB_CONN_MIN_EVICTABLE_IDLE_TIME_KEY,
-            ConfigurationKeys.DEFAULT_STATE_STORE_DB_CONN_MIN_EVICTABLE_IDLE_TIME));
 
-    return basicDataSource;
+    return dataSource;
   }
 
   /**

gobblin-metastore/src/main/java/org/apache/gobblin/metastore/MysqlStateStoreFactory.java

@@ -16,7 +16,7 @@
  */
 package org.apache.gobblin.metastore;
 
-import org.apache.commons.dbcp.BasicDataSource;
+import javax.sql.DataSource;
 
 import com.typesafe.config.Config;
 
@@ -36,10 +36,10 @@ public <T extends State> StateStore<T> createStateStore(Config config, Class<T>
             ConfigurationKeys.DEFAULT_STATE_STORE_COMPRESSED_VALUES);
 
     try {
-      BasicDataSource basicDataSource = MysqlDataSourceFactory.get(config,
+      DataSource dataSource = MysqlDataSourceFactory.get(config,
           SharedResourcesBrokerFactory.getImplicitBroker());
 
-      return new MysqlStateStore<>(basicDataSource, stateStoreTableName, compressedValues, stateClass);
+      return new MysqlStateStore<>(dataSource, stateStoreTableName, compressedValues, stateClass);
     } catch (Exception e) {
       throw new RuntimeException("Failed to create MysqlStateStore with factory", e);
     }