Merge branch 'master' into db2-tests

Author: rPraml

ebean-datasource-api/src/main/java/io/ebean/datasource/ConnectionPoolExhaustedException.java

@@ -0,0 +1,13 @@
+package io.ebean.datasource;
+
+import java.sql.SQLException;
+
+/**
+ * This exception is thrown, if the connection pool has reached maxSize.
+ * @author Roland Praml, Foconis Analytics GmbH
+ */
+public class ConnectionPoolExhaustedException extends SQLException {
+  public ConnectionPoolExhaustedException(String reason) {
+    super(reason);
+  }
+}

ebean-datasource-api/src/main/java/io/ebean/datasource/DataSourceBuilder.java

@@ -409,6 +409,15 @@ default DataSourceBuilder heartbeatTimeoutSeconds(int heartbeatTimeoutSeconds) {
   @Deprecated(forRemoval = true)
   DataSourceBuilder setHeartbeatTimeoutSeconds(int heartbeatTimeoutSeconds);
 
+  /**
+   * Sets the coun how often the heartbeat has to detect pool exhaustion in succession.
+   * in succession before an error is raised and the pool will be reset.
+   * <p>
+   * By default, this value must be multiplied with the sum of heartbeatfreq + waitTimeoutMillis to
+   * estimate the time, when the pool will be restarted, because all connections were leaked.
+   */
+  DataSourceBuilder heartbeatMaxPoolExhaustedCount(int count);
+
   /**
    * Set to true if a stack trace should be captured when obtaining a connection from the pool.
    * <p>
@@ -706,6 +715,7 @@ default DataSourceBuilder initDatabaseForPlatform(String platform) {
    * <p>
    * This is enabled by default. Generally we only want to turn this
    * off when using the pool with a Lambda function.
+   *
    * @param validateOnHeartbeat Use false to disable heartbeat validation.
    */
   DataSourceBuilder validateOnHeartbeat(boolean validateOnHeartbeat);
@@ -922,6 +932,11 @@ default String driverClassName() {
      */
     int getHeartbeatTimeoutSeconds();
 
+    /**
+     * Return the number, how often the heartbeat has to detect pool exhaustion in succession.
+     */
+    int getHeartbeatMaxPoolExhaustedCount();
+
     /**
      * Return true if a stack trace should be captured when obtaining a connection from the pool.
      * <p>

ebean-datasource-api/src/main/java/io/ebean/datasource/DataSourceConfig.java

@@ -62,6 +62,7 @@ public class DataSourceConfig implements DataSourceBuilder.Settings {
   private String heartbeatSql;
   private int heartbeatFreqSecs = 30;
   private int heartbeatTimeoutSeconds = 30;
+  private int heartbeatMaxPoolExhaustedCount = 10;
   private boolean captureStackTrace;
   private int maxStackTraceSize = 5;
   private int leakTimeMinutes = 30;
@@ -181,10 +182,10 @@ public DataSourceConfig setDefaults(DataSourceBuilder builder) {
     if (minConnections == 2 && other.getMinConnections() < 2) {
       minConnections = other.getMinConnections();
     }
-    if (!shutdownOnJvmExit && other.isShutdownOnJvmExit()){
+    if (!shutdownOnJvmExit && other.isShutdownOnJvmExit()) {
       shutdownOnJvmExit = true;
     }
-    if (validateOnHeartbeat && !other.isValidateOnHeartbeat()){
+    if (validateOnHeartbeat && !other.isValidateOnHeartbeat()) {
       validateOnHeartbeat = false;
     }
     if (customProperties == null) {
@@ -468,6 +469,17 @@ public DataSourceConfig setHeartbeatTimeoutSeconds(int heartbeatTimeoutSeconds)
     return this;
   }
 
+  @Override
+  public int getHeartbeatMaxPoolExhaustedCount() {
+    return this.heartbeatMaxPoolExhaustedCount;
+  }
+
+  @Override
+  public DataSourceBuilder heartbeatMaxPoolExhaustedCount(int count) {
+    this.heartbeatMaxPoolExhaustedCount = count;
+    return this;
+  }
+
   @Override
   public boolean isCaptureStackTrace() {
     return captureStackTrace;

ebean-datasource/src/main/java/io/ebean/datasource/pool/ConnectionPool.java

@@ -50,6 +50,8 @@ enum CloseWithinTxn {
   private final String heartbeatSql;
   private final int heartbeatFreqSecs;
   private final int heartbeatTimeoutSeconds;
+  private final int heartbeatMaxPoolExhaustedCount;
+
   private final long trimPoolFreqMillis;
   private final int transactionIsolation;
   private final boolean autoCommit;
@@ -82,6 +84,7 @@ enum CloseWithinTxn {
   private final int pstmtCacheSize;
   private final PooledConnectionQueue queue;
   private Timer heartBeatTimer;
+  private int heartbeatPoolExhaustedCount;
   /**
    * Used to find and close() leaked connections. Leaked connections are
    * thought to be busy but have not been used for some time. Each time a
@@ -118,6 +121,7 @@ enum CloseWithinTxn {
     this.waitTimeoutMillis = params.getWaitTimeoutMillis();
     this.heartbeatFreqSecs = params.getHeartbeatFreqSecs();
     this.heartbeatTimeoutSeconds = params.getHeartbeatTimeoutSeconds();
+    this.heartbeatMaxPoolExhaustedCount = params.getHeartbeatMaxPoolExhaustedCount();
     this.heartbeatSql = params.getHeartbeatSql();
     this.validateOnHeartbeat = params.isValidateOnHeartbeat();
     this.trimPoolFreqMillis = 1000L * params.getTrimPoolFreqSecs();
@@ -375,11 +379,19 @@ private void testConnection() {
     try {
       // Get a connection from the pool and test it
       conn = getConnection();
+      heartbeatPoolExhaustedCount = 0;
       if (testConnection(conn)) {
         notifyDataSourceIsUp();
       } else {
         notifyDataSourceIsDown(null);
       }
+    } catch (ConnectionPoolExhaustedException be) {
+      heartbeatPoolExhaustedCount++;
+      if (heartbeatPoolExhaustedCount > heartbeatMaxPoolExhaustedCount) {
+        notifyDataSourceIsDown(be);
+      } else {
+        Log.warn("Heartbeat: " + be.getMessage());
+      }
     } catch (SQLException ex) {
       notifyDataSourceIsDown(ex);
     } finally {
@@ -615,6 +627,7 @@ PooledConnection createConnectionForQueue(int connId) throws SQLException {
    * </ul>
    */
   private void reset() {
+    heartbeatPoolExhaustedCount = 0;
     queue.reset(leakTimeMinutes);
   }
 

ebean-datasource/src/main/java/io/ebean/datasource/pool/PooledConnection.java

@@ -250,6 +250,7 @@ void closeConnectionFully(boolean logErrors) {
         Log.error("Error checking if connection [" + name + "] is closed", ex);
       }
     }
+    lock.lock();
     try {
       for (ExtendedPreparedStatement ps : pstmtCache.values()) {
         ps.closeDestroy();
@@ -258,6 +259,19 @@ void closeConnectionFully(boolean logErrors) {
       if (logErrors) {
         Log.warn("Error when closing connection Statements", ex);
       }
+    } finally {
+      lock.unlock();
+    }
+    try {
+      // DB2 (and some other DBMS) may have uncommitted changes and do not allow close
+      // so try to do a rollback.
+      if (!connection.getAutoCommit()) {
+        connection.rollback();
+      }
+    } catch (SQLException ex) {
+      if (logErrors) {
+        Log.warn("Could not perform rollback", ex);
+      }
     }
     try {
       // DB2 (and some other DBMS) may have uncommitted changes and do not allow close

ebean-datasource/src/main/java/io/ebean/datasource/pool/PooledConnectionQueue.java

@@ -1,5 +1,6 @@
 package io.ebean.datasource.pool;
 
+import io.ebean.datasource.ConnectionPoolExhaustedException;
 import io.ebean.datasource.PoolStatus;
 import io.ebean.datasource.pool.ConnectionPool.Status;
 
@@ -226,18 +227,13 @@ private PooledConnection _obtainConnection() throws InterruptedException, SQLExc
       hitCount++;
       // are other threads already waiting? (they get priority)
       if (waitingThreads == 0) {
-        PooledConnection freeConnection = extractFromFreeList();
-        if (freeConnection != null) {
-          return freeConnection;
+        PooledConnection connection = extractFromFreeList();
+        if (connection != null) {
+          return connection;
         }
-        if (busyList.size() < maxSize) {
-          // grow the connection pool
-          PooledConnection c = pool.createConnectionForQueue(connectionId++);
-          int busySize = registerBusyConnection(c);
-          if (Log.isLoggable(DEBUG)) {
-            Log.debug("DataSource [{0}] grow; id[{1}] busy[{2}] max[{3}]", name, c.name(), busySize, maxSize);
-          }
-          return c;
+        connection = createConnection();
+        if (connection != null) {
+          return connection;
         }
       }
       try {
@@ -257,21 +253,40 @@ private PooledConnection _obtainConnection() throws InterruptedException, SQLExc
     }
   }
 
+  private PooledConnection createConnection() throws SQLException {
+    if (busyList.size() < maxSize) {
+      // grow the connection pool
+      PooledConnection c = pool.createConnectionForQueue(connectionId++);
+      int busySize = registerBusyConnection(c);
+      if (Log.isLoggable(DEBUG)) {
+        Log.debug("DataSource [{0}] grow; id[{1}] busy[{2}] max[{3}]", name, c.name(), busySize, maxSize);
+      }
+      return c;
+    } else {
+      return null;
+    }
+  }
+
   /**
    * Got into a loop waiting for connections to be returned to the pool.
    */
   private PooledConnection _obtainConnectionWaitLoop() throws SQLException, InterruptedException {
     long nanos = MILLIS_TIME_UNIT.toNanos(waitTimeoutMillis);
     for (; ; ) {
       if (nanos <= 0) {
+        // We waited long enough, that a connection was returned, so we try to create a new connection.
+        PooledConnection conn = createConnection();
+        if (conn != null) {
+          return conn;
+        }
         String msg = "Unsuccessfully waited [" + waitTimeoutMillis + "] millis for a connection to be returned."
           + " No connections are free. You need to Increase the max connections of [" + maxSize + "]"
           + " or look for a connection pool leak using datasource.xxx.capturestacktrace=true";
         if (pool.captureStackTrace()) {
           dumpBusyConnectionInformation();
         }
 
-        throw new SQLException(msg);
+        throw new ConnectionPoolExhaustedException(msg);
       }
 
       try {

ebean-datasource/src/test/java/io/ebean/datasource/pool/ConnectionPoolFullTest.java

@@ -0,0 +1,88 @@
+package io.ebean.datasource.pool;
+
+import io.ebean.datasource.DataSourceAlert;
+import io.ebean.datasource.DataSourceBuilder;
+import io.ebean.datasource.DataSourcePool;
+import org.junit.jupiter.api.Test;
+
+import javax.sql.DataSource;
+import java.sql.Connection;
+import java.sql.SQLException;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class ConnectionPoolFullTest implements DataSourceAlert {
+
+  private int up;
+  private int down;
+
+  @Test
+  void testPoolFullWithHeartbeat() throws Exception {
+
+    DataSourcePool pool = DataSourceBuilder.create()
+      .url("jdbc:h2:mem:testConnectionPoolFull")
+      .username("sa")
+      .password("sa")
+      .heartbeatFreqSecs(1)
+      .minConnections(1)
+      .maxConnections(1)
+      .trimPoolFreqSecs(1)
+      .heartbeatMaxPoolExhaustedCount(1)
+      .alert(this)
+      .failOnStart(false)
+      .build();
+
+    assertThat(up).isEqualTo(1);
+    assertThat(down).isEqualTo(0);
+
+    try {
+      // block the thread for 2 secs. The heartbeat must not shutdown the pool
+      try (Connection connection = pool.getConnection()) {
+        System.out.println("waiting 2s");
+        Thread.sleep(2000);
+        connection.rollback();
+      }
+      assertThat(up).isEqualTo(1);
+      assertThat(down).isEqualTo(0);
+
+      // now block the thread longer, so that exhausted count will be reached
+      try (Connection connection = pool.getConnection()) {
+        System.out.println("waiting 4s");
+        Thread.sleep(4000);
+        connection.rollback();
+      }
+      // we expect, that the pool goes down.
+      assertThat(up).isEqualTo(1);
+      assertThat(down).isEqualTo(1);
+
+      System.out.println("waiting 2s for recovery");
+      Thread.sleep(2000);
+      assertThat(up).isEqualTo(2);
+      assertThat(down).isEqualTo(1);
+
+      // pool should be OK again
+      try (Connection connection = pool.getConnection()) {
+        connection.rollback();
+      }
+
+      assertThat(up).isEqualTo(2);
+      assertThat(down).isEqualTo(1);
+
+
+    } finally {
+      pool.shutdown();
+    }
+
+  }
+
+
+  @Override
+  public void dataSourceUp(DataSource dataSource) {
+    up++;
+  }
+
+  @Override
+  public void dataSourceDown(DataSource dataSource, SQLException reason) {
+    down++;
+  }
+}

ebean-datasource/src/test/java/io/ebean/datasource/pool/ConnectionPoolRecoverTest.java

@@ -0,0 +1,43 @@
+package io.ebean.datasource.pool;
+
+import io.ebean.datasource.DataSourceBuilder;
+import io.ebean.datasource.DataSourcePool;
+import org.junit.jupiter.api.Test;
+
+import java.sql.Connection;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class ConnectionPoolRecoverTest {
+
+  @Test
+  void testHeavyLoadPool() throws Exception {
+    DataSourcePool pool = DataSourceBuilder.create()
+      .url("jdbc:h2:mem:testConnectionPoolFull")
+      .username("sa")
+      .password("sa")
+      .heartbeatFreqSecs(1)
+      .minConnections(1)
+      .maxConnections(1)
+      .trimPoolFreqSecs(1)
+      //     .heartbeatMaxPoolExhaustedCount(1)
+      .failOnStart(false)
+      .build();
+    try {
+      for (int i = 0; i < 5; i++) {
+        try (Connection conn = pool.getConnection()) {
+          Thread.sleep(2000);
+          conn.rollback();
+        }
+      }
+    } finally {
+      pool.shutdown();
+    }
+  }
+
+}