@SuppressWarnings({ "rawtypes", "unchecked" })
  @Override
  public DataSource create(Configuration hdpConfig, int maxPoolSize) throws SQLException {
    String poolName = DataSourceProvider.getDataSourceName(hdpConfig);
    LOG.info("Creating dbcp connection pool for the MetaStore, maxPoolSize: {}, name: {}", maxPoolSize, poolName);

    String driverUrl = DataSourceProvider.getMetastoreJdbcDriverUrl(hdpConfig);
    String user = DataSourceProvider.getMetastoreJdbcUser(hdpConfig);
    String passwd = DataSourceProvider.getMetastoreJdbcPasswd(hdpConfig);

    BasicDataSource dbcpDs = new BasicDataSource();
    dbcpDs.setUrl(driverUrl);
    dbcpDs.setUsername(user);
    dbcpDs.setPassword(passwd);
    dbcpDs.setDefaultReadOnly(false);
    dbcpDs.setDefaultAutoCommit(true);

    DatabaseProduct dbProduct =  DatabaseProduct.determineDatabaseProduct(driverUrl, hdpConfig);
    Map<String, String> props = dbProduct.getDataSourceProperties();
    for (Map.Entry<String, String> kv : props.entrySet()) {
      dbcpDs.setConnectionProperties(kv.getKey() + "=" + kv.getValue());
    }

    long connectionTimeout = hdpConfig.getLong(CONNECTION_TIMEOUT_PROPERTY, 30000L);
    int connectionMaxIlde = hdpConfig.getInt(CONNECTION_MAX_IDLE_PROPERTY, 8);
    int connectionMinIlde = hdpConfig.getInt(CONNECTION_MIN_IDLE_PROPERTY, 0);
    boolean testOnBorrow = hdpConfig.getBoolean(CONNECTION_TEST_BORROW_PROPERTY,
        BaseObjectPoolConfig.DEFAULT_TEST_ON_BORROW);
    long evictionTimeMillis = hdpConfig.getLong(CONNECTION_MIN_EVICT_MILLIS_PROPERTY,
        BaseObjectPoolConfig.DEFAULT_MIN_EVICTABLE_IDLE_DURATION.toMillis());
    boolean testWhileIdle = hdpConfig.getBoolean(CONNECTION_TEST_IDLEPROPERTY,
        BaseObjectPoolConfig.DEFAULT_TEST_WHILE_IDLE);
    long timeBetweenEvictionRuns = hdpConfig.getLong(CONNECTION_TIME_BETWEEN_EVICTION_RUNS_MILLIS,
        BaseObjectPoolConfig.DEFAULT_DURATION_BETWEEN_EVICTION_RUNS.toMillis());
    int numTestsPerEvictionRun = hdpConfig.getInt(CONNECTION_NUM_TESTS_PER_EVICTION_RUN,
        BaseObjectPoolConfig.DEFAULT_NUM_TESTS_PER_EVICTION_RUN);
    boolean testOnReturn = hdpConfig.getBoolean(CONNECTION_TEST_ON_RETURN, BaseObjectPoolConfig.DEFAULT_TEST_ON_RETURN);
    long softMinEvictableIdleTimeMillis = hdpConfig.getLong(CONNECTION_SOFT_MIN_EVICTABLE_IDLE_TIME,
        BaseObjectPoolConfig.DEFAULT_SOFT_MIN_EVICTABLE_IDLE_DURATION.toMillis());
    boolean lifo = hdpConfig.getBoolean(CONNECTION_LIFO, BaseObjectPoolConfig.DEFAULT_LIFO);

    ConnectionFactory connFactory = new DataSourceConnectionFactory(dbcpDs);
    PoolableConnectionFactory poolableConnFactory = new PoolableConnectionFactory(connFactory, null);

    GenericObjectPool objectPool = new GenericObjectPool(poolableConnFactory);
    objectPool.setMaxTotal(maxPoolSize);
    objectPool.setMaxWait(Duration.ofMillis(connectionTimeout));
    objectPool.setMaxIdle(connectionMaxIlde);
    objectPool.setMinIdle(connectionMinIlde);
    objectPool.setTestOnBorrow(testOnBorrow);
    objectPool.setTestWhileIdle(testWhileIdle);
    objectPool.setMinEvictableIdleDuration(Duration.ofMillis(evictionTimeMillis));
    objectPool.setDurationBetweenEvictionRuns(Duration.ofMillis(timeBetweenEvictionRuns));
    objectPool.setNumTestsPerEvictionRun(numTestsPerEvictionRun);
    objectPool.setTestOnReturn(testOnReturn);
    objectPool.setSoftMinEvictableIdleDuration(Duration.ofMillis(softMinEvictableIdleTimeMillis));
    objectPool.setLifo(lifo);

    // Enable TxnHandler#connPoolMutex to release the idle connection if possible,
    // TxnHandler#connPoolMutex is mostly used for MutexAPI that is primarily designed to
    // provide coarse-grained mutex support to maintenance tasks running inside the Metastore,
    // this will make Metastore more scalable especially if there is a leader in the warehouse.
    if ("mutex".equalsIgnoreCase(poolName)) {
      if (timeBetweenEvictionRuns < 0) {
        // When timeBetweenEvictionRunsMillis non-positive, no idle object evictor thread runs
        objectPool.setDurationBetweenEvictionRuns(Duration.ofMillis(30 * 1000));
      }
      if (softMinEvictableIdleTimeMillis < 0) {
        objectPool.setSoftMinEvictableIdleDuration(Duration.ofMillis(600 * 1000));
      }
    }
    String stmt = dbProduct.getPrepareTxnStmt();
    if (stmt != null) {
      poolableConnFactory.setConnectionInitSql(Collections.singletonList(stmt));
    }
    return new PoolingDataSource(objectPool);
  }
DataSourceProvider for the dbcp connection pool.
standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/datasource/DbCPDataSourceProvider.java