BITMAG-1248 Replace java.util.Timer with ScheduledExecutorService + virtual threads

bitrepository-audit-trail-service/src/main/java/org/bitrepository/audittrails/collector/AuditTrailCollectionTimerTask.java

@@ -28,9 +28,8 @@
 
 import java.time.Instant;
 import java.util.Date;
-import java.util.TimerTask;
 
-public class AuditTrailCollectionTimerTask extends TimerTask {
+public class AuditTrailCollectionTimerTask implements Runnable {
     private final Logger log = LoggerFactory.getLogger(getClass());
 
     private final IncrementalCollector collector;

bitrepository-audit-trail-service/src/main/java/org/bitrepository/audittrails/collector/AuditTrailCollector.java

@@ -29,6 +29,7 @@
 import org.bitrepository.audittrails.store.AuditTrailStore;
 import org.bitrepository.audittrails.webservice.CollectorInfo;
 import org.bitrepository.common.ArgumentValidator;
+import org.bitrepository.common.ScheduledVirtualThreadExecutor;
 import org.bitrepository.common.settings.Settings;
 import org.bitrepository.common.utils.SettingsUtils;
 import org.bitrepository.common.utils.TimeUtils;
@@ -42,15 +43,15 @@
 import java.time.Instant;
 import java.util.HashMap;
 import java.util.Map;
-import java.util.Timer;
+import java.util.concurrent.TimeUnit;
 
 /**
  * Manages the retrieval of AuditTrails from contributors.
  */
 public class AuditTrailCollector extends AuditTrailTaskStarter {
     private final Logger log = LoggerFactory.getLogger(getClass());
     private final Map<String, AuditTrailCollectionTimerTask> collectorTasks = new HashMap<>();
-    private final Timer timer;
+    private final ScheduledVirtualThreadExecutor scheduler;
 
     /**
      * @param settings        The settings for this collector.
@@ -64,7 +65,7 @@ public AuditTrailCollector(Settings settings, AuditTrailClient client, AuditTrai
         ArgumentValidator.checkNotNull(client, "AuditTrailClient client");
         ArgumentValidator.checkNotNull(alarmDispatcher, "AlarmDispatcher alarmDispatcher");
 
-        this.timer = new Timer(true);
+        this.scheduler = new ScheduledVirtualThreadExecutor("AuditTrailCollector", true);
         javax.xml.datatype.Duration collectAuditInterval =
                 settings.getReferenceSettings().getAuditTrailServiceSettings().getCollectAuditInterval();
         Duration collectionInterval = XmlUtils.xmlDurationToDuration(collectAuditInterval);
@@ -80,7 +81,8 @@ public AuditTrailCollector(Settings settings, AuditTrailClient client, AuditTrai
                     collector, collectionInterval.toMillis(), Math.toIntExact(collectionGracePeriod.toMillis()));
             log.info("Starting collection of audit trails every {} after grace period of {}.",
                     TimeUtils.durationToHuman(collectionInterval), TimeUtils.durationToHuman(collectionGracePeriod));
-            timer.scheduleAtFixedRate(collectorTask, collectionGracePeriod.toMillis(), collectionInterval.toMillis());
+            scheduler.scheduleAtFixedRate(collectorTask, collectionGracePeriod.toMillis(), collectionInterval.toMillis(),
+                    TimeUnit.MILLISECONDS);
             collectorTasks.put(c.getID(), collectorTask);
         }
     }
@@ -123,10 +125,7 @@ public void collectNewestAudits(String collectionID) {
      * Closes the AuditTrailCollector.
      */
     public void close() {
-        for (AuditTrailCollectionTimerTask timerTask : collectorTasks.values()) {
-            timerTask.cancel();
-        }
-        timer.cancel();
+        scheduler.close();
     }
 
 }

bitrepository-audit-trail-service/src/main/java/org/bitrepository/audittrails/preserver/LocalAuditTrailPreserver.java

@@ -29,6 +29,7 @@
 import org.bitrepository.bitrepositoryelements.ChecksumSpecTYPE;
 import org.bitrepository.client.eventhandler.EventHandler;
 import org.bitrepository.common.ArgumentValidator;
+import org.bitrepository.common.ScheduledVirtualThreadExecutor;
 import org.bitrepository.common.TimerTaskSchedule;
 import org.bitrepository.common.exceptions.OperationFailedException;
 import org.bitrepository.common.settings.Settings;
@@ -58,8 +59,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.Timer;
-import java.util.TimerTask;
+import java.util.concurrent.TimeUnit;
 
 /**
  * Handles the preservation of audit trails to a collection defined for the local repository.
@@ -71,7 +71,7 @@ public class LocalAuditTrailPreserver extends AuditTrailTaskStarter implements A
     private final Map<String, AuditPacker> auditPackers = new HashMap<>();
     private final AuditTrailPreservation preservationSettings;
     private final FileExchange exchange;
-    private Timer timer;
+    private ScheduledVirtualThreadExecutor scheduler;
     private AuditPreservationTimerTask preservationTask = null;
     private long preservedAuditCount = 0;
 
@@ -114,27 +114,27 @@ private void initializeDatabaseEntries() {
 
     @Override
     public void start() {
-        if (timer != null) {
-            log.debug("Cancelling old timer.");
-            timer.cancel();
+        if (scheduler != null) {
+            log.debug("Cancelling old scheduler.");
+            scheduler.close();
         }
 
         javax.xml.datatype.Duration preservationIntervalXmlDur = preservationSettings.getAuditTrailPreservationInterval();
         Duration preservationInterval = XmlUtils.xmlDurationToDuration(preservationIntervalXmlDur);
         Duration preservationGracePeriod = getGracePeriod();
         log.info("Starting preservation of audit trails every {} after grace period of {}.",
                 TimeUtils.durationToHuman(preservationInterval), TimeUtils.durationToHuman(preservationGracePeriod));
-        timer = new Timer(true);
+        scheduler = new ScheduledVirtualThreadExecutor("AuditTrailPreserver", true);
         preservationTask = new AuditPreservationTimerTask(preservationInterval.toMillis(),
                 Math.toIntExact(preservationGracePeriod.toMillis()));
-        timer.scheduleAtFixedRate(preservationTask, preservationGracePeriod.toMillis(), preservationInterval.toMillis());
+        scheduler.scheduleAtFixedRate(preservationTask, preservationGracePeriod.toMillis(), preservationInterval.toMillis(),
+                TimeUnit.MILLISECONDS);
     }
 
     @Override
     public void close() {
-        if (timer != null) {
-            preservationTask.cancel();
-            timer.cancel();
+        if (scheduler != null) {
+            scheduler.close();
         }
     }
 
@@ -260,7 +260,7 @@ public PreservationInfo getPreservationInfo() {
     /**
      * Timer task for keeping track of the automated collecting of audit trails.
      */
-    private class AuditPreservationTimerTask extends TimerTask {
+    private class AuditPreservationTimerTask implements Runnable {
         private final Logger log = LoggerFactory.getLogger(getClass());
         private final TimerTaskSchedule schedule;
 

bitrepository-client/src/main/java/org/bitrepository/client/conversation/mediator/CollectionBasedConversationMediator.java

@@ -28,6 +28,7 @@
 import org.bitrepository.client.conversation.Conversation;
 import org.bitrepository.client.eventhandler.OperationFailedEvent;
 import org.bitrepository.common.DefaultThreadFactory;
+import org.bitrepository.common.ScheduledVirtualThreadExecutor;
 import org.bitrepository.common.settings.Settings;
 import org.bitrepository.common.utils.TimeUtils;
 import org.bitrepository.common.utils.XmlUtils;
@@ -43,9 +44,9 @@
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
-import java.util.Timer;
-import java.util.TimerTask;
+import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
 
 /**
  * Conversation handler that delegates messages to registered conversations.
@@ -61,22 +62,27 @@ public class CollectionBasedConversationMediator implements ConversationMediator
      *
      * @see ConversationCleaner
      */
-    private static final Timer cleanTimer = new Timer(NAME_OF_TIMER, TIMER_IS_DAEMON);
+    private static final ScheduledVirtualThreadExecutor cleanScheduler =
+            new ScheduledVirtualThreadExecutor(NAME_OF_TIMER, TIMER_IS_DAEMON);
     private final MessageBus messagebus;
     private static final ThreadFactory threadFactory = new DefaultThreadFactory(
             CollectionBasedConversationMediator.class.getSimpleName() + "-", Thread.NORM_PRIORITY, false);
+    private ScheduledFuture<?> cleanupTask;
 
     @Override
     public void start() {
         messagebus.addListener(settings.getReceiverDestinationID(), this);
         javax.xml.datatype.Duration cleanupInterval = settings.getReferenceSettings().getClientSettings().getMediatorCleanupInterval();
-        cleanTimer.scheduleAtFixedRate(new ConversationCleaner(),
-                0, XmlUtils.xmlDurationToMilliseconds(cleanupInterval));
+        cleanupTask = cleanScheduler.scheduleAtFixedRate(new ConversationCleaner(),
+                0, XmlUtils.xmlDurationToMilliseconds(cleanupInterval), TimeUnit.MILLISECONDS);
     }
 
     @Override
     public void shutdown() {
         messagebus.removeListener(settings.getReceiverDestinationID(), this);
+        if (cleanupTask != null) {
+            cleanupTask.cancel(false);
+        }
     }
 
     /**
@@ -149,7 +155,7 @@ public void onMessage(Message message, MessageContext messageContext) {
      * A copy of the current conversations is created before running through the conversations to avoid having to lock
      * the conversations map while cleaning.
      */
-    private final class ConversationCleaner extends TimerTask {
+    private final class ConversationCleaner implements Runnable {
         @Override
         public void run() {
             Conversation[] conversationArray = conversations.values().toArray(new Conversation[0]);

bitrepository-core/src/main/java/org/bitrepository/common/ScheduledVirtualThreadExecutor.java

@@ -0,0 +1,101 @@
+/*
+ * #%L
+ * Bitrepository Protocol
+ *
+ * $Id$
+ * $HeadURL$
+ * %%
+ * Copyright (C) 2010 - 2011 The Royal Library and The State Archives, Denmark
+ * %%
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Lesser General Public License as
+ * published by the Free Software Foundation, either version 2.1 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Lesser Public License for more details.
+ *
+ * You should have received a copy of the GNU General Lesser Public
+ * License along with this program.  If not, see
+ * <http://www.gnu.org/licenses/lgpl-2.1.html>.
+ * #L%
+ */
+package org.bitrepository.common;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Schedules recurring or delayed tasks the way {@link java.util.Timer} used to, but dispatches each firing to its
+ * own virtual thread instead of running it on the scheduling thread itself. A single-thread, named platform-thread
+ * {@link ScheduledExecutorService} acts purely as the "ticker" - it never runs task bodies itself, so a slow or
+ * blocking task can't delay the next scheduled firing, and the ticker thread is never split up amongst virtual
+ * threads.
+ * <p>
+ * Uncaught exceptions escaping a dispatched task are logged and otherwise ignored, so one failing run doesn't stop
+ * future runs - unlike {@link java.util.Timer}, where an uncaught exception silently kills the whole timer thread.
+ */
+public class ScheduledVirtualThreadExecutor implements AutoCloseable {
+    private final ScheduledExecutorService ticker;
+    private final ExecutorService worker;
+
+    /**
+     * @param name   Prefix used for both the ticker thread's name and the dispatched virtual threads' names.
+     * @param daemon Whether the ticker thread should be a daemon thread.
+     */
+    public ScheduledVirtualThreadExecutor(String name, boolean daemon) {
+        ticker = Executors.newSingleThreadScheduledExecutor(new DefaultThreadFactory(name, Thread.NORM_PRIORITY, daemon));
+        worker = Executors.newThreadPerTaskExecutor(Thread.ofVirtual().name(name + "-worker-", 0)
+                .uncaughtExceptionHandler(ScheduledVirtualThreadExecutor::logUncaughtException)
+                .factory());
+    }
+
+    private static void logUncaughtException(Thread thread, Throwable throwable) {
+        String throwingClass = throwable.getStackTrace()[0].getClassName();
+        Logger logger = LoggerFactory.getLogger(throwingClass);
+        logger.error("UncaughtExceptionHandler caught Exception:", throwable);
+    }
+
+    /**
+     * Dispatches {@code command} to a new virtual thread once, after {@code delay}.
+     */
+    public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) {
+        return ticker.schedule(() -> worker.execute(command), delay, unit);
+    }
+
+    /**
+     * Dispatches {@code command} to a new virtual thread every {@code period}, starting after {@code initialDelay}.
+     * Ticks continue at the fixed rate regardless of how long a dispatched run takes, matching
+     * {@link java.util.Timer#scheduleAtFixedRate}.
+     */
+    public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit) {
+        return ticker.scheduleAtFixedRate(() -> worker.execute(command), initialDelay, period, unit);
+    }
+
+    /**
+     * Dispatches {@code command} to a new virtual thread, waiting {@code delay} after each dispatch before
+     * scheduling the next one, starting after {@code initialDelay}. Matches {@link java.util.Timer#schedule}'s
+     * fixed-delay semantics.
+     */
+    public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit) {
+        return ticker.scheduleWithFixedDelay(() -> worker.execute(command), initialDelay, delay, unit);
+    }
+
+    /**
+     * Stops the ticker from scheduling further work and stops accepting new dispatches. Tasks already dispatched
+     * to a virtual thread are left to finish on their own.
+     */
+    @Override
+    public void close() {
+        ticker.shutdownNow();
+        worker.shutdown();
+    }
+}