Internal heartbeat Timer, use Virtual Thread for Java 21

Using Multi-Version jar, use Virtual Thread instead of Timer for executing the heartbeat test

Author: rbygrave

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

@@ -26,6 +26,12 @@
  */
 final class ConnectionPool implements DataSourcePool {
 
+  @FunctionalInterface
+  interface Heartbeat {
+
+    void stop();
+  }
+
   private static final String APPLICATION_NAME = "ApplicationName";
   private final ReentrantLock heartbeatLock = new ReentrantLock(false);
   private final ReentrantLock notifyLock = new ReentrantLock(false);
@@ -80,7 +86,7 @@ final class ConnectionPool implements DataSourcePool {
   private final int waitTimeoutMillis;
   private final int pstmtCacheSize;
   private final PooledConnectionQueue queue;
-  private Timer heartBeatTimer;
+  private Heartbeat heartbeat;
   private int heartbeatPoolExhaustedCount;
   private final ExecutorService executor;
 
@@ -161,13 +167,6 @@ void pstmtCacheMetrics(PstmtCache pstmtCache) {
     pscRem.add(pstmtCache.removeCount());
   }
 
-  final class HeartBeatRunnable extends TimerTask {
-    @Override
-    public void run() {
-      heartBeat();
-    }
-  }
-
   @Override
   public java.util.logging.Logger getParentLogger() throws SQLFeatureNotSupportedException {
     throw new SQLFeatureNotSupportedException("We do not support java.util.logging");
@@ -387,7 +386,7 @@ private void trimIdleConnections() {
    * This is called by the HeartbeatRunnable which should be scheduled to
    * run periodically (every heartbeatFreqSecs seconds).
    */
-  private void heartBeat() {
+  void heartbeat() {
     trimIdleConnections();
     if (validateOnHeartbeat) {
       testConnection();
@@ -727,11 +726,10 @@ private void startHeartBeatIfStopped() {
     heartbeatLock.lock();
     try {
       // only start if it is not already running
-      if (heartBeatTimer == null) {
+      if (heartbeat == null) {
         int freqMillis = heartbeatFreqSecs * 1000;
         if (freqMillis > 0) {
-          heartBeatTimer = new Timer(name + ".heartBeat", true);
-          heartBeatTimer.scheduleAtFixedRate(new HeartBeatRunnable(), freqMillis, freqMillis);
+          heartbeat = ExecutorFactory.newHeartBeat(this, freqMillis);
         }
       }
     } finally {
@@ -743,9 +741,9 @@ private void stopHeartBeatIfRunning() {
     heartbeatLock.lock();
     try {
       // only stop if it was running
-      if (heartBeatTimer != null) {
-        heartBeatTimer.cancel();
-        heartBeatTimer = null;
+      if (heartbeat != null) {
+        heartbeat.stop();
+        heartbeat = null;
       }
     } finally {
       heartbeatLock.unlock();

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

@@ -1,5 +1,9 @@
 package io.ebean.datasource.pool;
 
+import io.ebean.datasource.pool.ConnectionPool.Heartbeat;
+
+import java.util.Timer;
+import java.util.TimerTask;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 
@@ -8,4 +12,41 @@ final class ExecutorFactory {
   static ExecutorService newExecutor() {
     return Executors.newSingleThreadExecutor();
   }
+
+  /**
+   * Return a new Heartbeat for the pool.
+   */
+  static Heartbeat newHeartBeat(ConnectionPool pool, int freqMillis) {
+    final Timer timer = new Timer(pool.name() + ".heartbeat", true);
+    timer.scheduleAtFixedRate(new HeartbeatTask(pool), freqMillis, freqMillis);
+    return new TimerHeartbeat(timer);
+  }
+
+  private static final class TimerHeartbeat implements Heartbeat {
+
+    private final Timer timer;
+
+    private TimerHeartbeat(Timer timer) {
+      this.timer = timer;
+    }
+
+    @Override
+    public void stop() {
+      timer.cancel();
+    }
+  }
+
+  private static final class HeartbeatTask extends TimerTask {
+
+    private final ConnectionPool pool;
+
+    private HeartbeatTask(ConnectionPool pool) {
+      this.pool = pool;
+    }
+
+    @Override
+    public void run() {
+      pool.heartbeat();
+    }
+  }
 }

ebean-datasource/src/main/java21/io/ebean/datasource/pool/ExecutorFactory.java

@@ -1,11 +1,61 @@
 package io.ebean.datasource.pool;
 
+import io.ebean.datasource.pool.ConnectionPool.Heartbeat;
+
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 final class ExecutorFactory {
 
   static ExecutorService newExecutor() {
     return Executors.newVirtualThreadPerTaskExecutor();
   }
+
+  static Heartbeat newHeartBeat(ConnectionPool pool, int freqMillis) {
+    return new VTHeartbeat(pool, freqMillis).start();
+  }
+
+  private static final class VTHeartbeat implements Heartbeat {
+
+    private final AtomicBoolean running = new AtomicBoolean(false);
+    private final ConnectionPool pool;
+    private final int freqMillis;
+    private final Thread thread;
+
+    private VTHeartbeat(ConnectionPool pool, int freqMillis) {
+      this.pool = pool;
+      this.freqMillis = freqMillis;
+      this.thread = Thread.ofVirtual()
+              .name(pool.name() + ".heartbeat")
+              .unstarted(this::run);
+    }
+
+    private void run() {
+      while (running.get()) {
+        try {
+          Thread.sleep(freqMillis);
+          pool.heartbeat();
+        } catch (InterruptedException e) {
+          Thread.currentThread().interrupt();
+          break;
+        } catch (Exception e) {
+          // continue heartbeat
+          Log.warn("Error during heartbeat", e);
+        }
+      }
+    }
+
+    private Heartbeat start() {
+      running.set(true);
+      thread.start();
+      return this;
+    }
+
+    @Override
+    public void stop() {
+      running.set(false);
+      thread.interrupt();
+    }
+  }
 }