Merge pull request #111 from FOCONIS/heartbeat-when-cp-is-full

Heartbeat should not reset, when pool is full

Author: rbygrave

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);
@@ -900,6 +910,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;
@@ -179,10 +180,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) {
@@ -466,6 +467,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

@@ -46,6 +46,8 @@ final class ConnectionPool implements DataSourcePool {
   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;
@@ -77,6 +79,7 @@ final class ConnectionPool implements DataSourcePool {
   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
@@ -112,6 +115,7 @@ final class ConnectionPool implements DataSourcePool {
     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();
@@ -367,11 +371,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 {
@@ -576,6 +588,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/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;
 
@@ -271,7 +272,7 @@ private PooledConnection _obtainConnectionWaitLoop() throws SQLException, Interr
           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++;
+  }
+}