diff --git a/bitrepository-audit-trail-service/src/main/java/org/bitrepository/audittrails/collector/AuditTrailCollectionTimerTask.java b/bitrepository-audit-trail-service/src/main/java/org/bitrepository/audittrails/collector/AuditTrailCollectionTimerTask.java index e089dd407..daf2155fd 100644 --- a/bitrepository-audit-trail-service/src/main/java/org/bitrepository/audittrails/collector/AuditTrailCollectionTimerTask.java +++ b/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; diff --git a/bitrepository-audit-trail-service/src/main/java/org/bitrepository/audittrails/collector/AuditTrailCollector.java b/bitrepository-audit-trail-service/src/main/java/org/bitrepository/audittrails/collector/AuditTrailCollector.java index f4756ebb3..b80d12315 100644 --- a/bitrepository-audit-trail-service/src/main/java/org/bitrepository/audittrails/collector/AuditTrailCollector.java +++ b/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,7 +43,7 @@ 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. @@ -50,7 +51,7 @@ public class AuditTrailCollector extends AuditTrailTaskStarter { private final Logger log = LoggerFactory.getLogger(getClass()); private final Map 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(); } } diff --git a/bitrepository-audit-trail-service/src/main/java/org/bitrepository/audittrails/preserver/LocalAuditTrailPreserver.java b/bitrepository-audit-trail-service/src/main/java/org/bitrepository/audittrails/preserver/LocalAuditTrailPreserver.java index c66e9167f..e5a22c095 100644 --- a/bitrepository-audit-trail-service/src/main/java/org/bitrepository/audittrails/preserver/LocalAuditTrailPreserver.java +++ b/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 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,9 +114,9 @@ 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(); @@ -124,17 +124,17 @@ public void start() { 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; diff --git a/bitrepository-client/src/main/java/org/bitrepository/client/conversation/mediator/CollectionBasedConversationMediator.java b/bitrepository-client/src/main/java/org/bitrepository/client/conversation/mediator/CollectionBasedConversationMediator.java index 4e57381d9..060c56e81 100644 --- a/bitrepository-client/src/main/java/org/bitrepository/client/conversation/mediator/CollectionBasedConversationMediator.java +++ b/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]); diff --git a/bitrepository-core/src/main/java/org/bitrepository/common/ScheduledVirtualThreadExecutor.java b/bitrepository-core/src/main/java/org/bitrepository/common/ScheduledVirtualThreadExecutor.java new file mode 100644 index 000000000..89d22582f --- /dev/null +++ b/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 + * . + * #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. + *

+ * 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(); + } +} diff --git a/bitrepository-core/src/test/java/org/bitrepository/common/ScheduledVirtualThreadExecutorTest.java b/bitrepository-core/src/test/java/org/bitrepository/common/ScheduledVirtualThreadExecutorTest.java new file mode 100644 index 000000000..b57b46780 --- /dev/null +++ b/bitrepository-core/src/test/java/org/bitrepository/common/ScheduledVirtualThreadExecutorTest.java @@ -0,0 +1,206 @@ +/* + * #%L + * Bitrepository Protocol + * + * $Id$ + * $HeadURL$ + * %% + * Copyright (C) 2010 - 2026 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 + * . + * #L% + */ +package org.bitrepository.common; + +import ch.qos.logback.classic.Level; +import ch.qos.logback.classic.spi.ILoggingEvent; +import ch.qos.logback.core.Appender; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; +import org.slf4j.LoggerFactory; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.bitrepository.common.utils.AllureTestUtils.addDescription; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.timeout; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +class ScheduledVirtualThreadExecutorTest { + private ScheduledVirtualThreadExecutor executor; + + @AfterEach + void tearDown() { + if (executor != null) { + executor.close(); + } + } + + @Test + @Tag("regressiontest") + void scheduleAtFixedRateRunsRepeatedlyTest() throws Exception { + addDescription("Test that scheduleAtFixedRate dispatches the task repeatedly."); + executor = new ScheduledVirtualThreadExecutor("fixedRateTest", true); + CountDownLatch latch = new CountDownLatch(3); + + executor.scheduleAtFixedRate(latch::countDown, 0, 20, TimeUnit.MILLISECONDS); + + Assertions.assertTrue(latch.await(5, TimeUnit.SECONDS), "Task should have run at least 3 times"); + } + + @Test + @Tag("regressiontest") + void scheduleWithFixedDelayRunsRepeatedlyTest() throws Exception { + addDescription("Test that scheduleWithFixedDelay dispatches the task repeatedly."); + executor = new ScheduledVirtualThreadExecutor("fixedDelayTest", true); + CountDownLatch latch = new CountDownLatch(3); + + executor.scheduleWithFixedDelay(latch::countDown, 0, 20, TimeUnit.MILLISECONDS); + + Assertions.assertTrue(latch.await(5, TimeUnit.SECONDS), "Task should have run at least 3 times"); + } + + @Test + @Tag("regressiontest") + void scheduleRunsOnlyOnceTest() throws Exception { + addDescription("Test that a one-shot schedule() call dispatches the task exactly once."); + executor = new ScheduledVirtualThreadExecutor("onceTest", true); + AtomicInteger count = new AtomicInteger(); + CountDownLatch latch = new CountDownLatch(1); + + executor.schedule(() -> { + count.incrementAndGet(); + latch.countDown(); + }, 0, TimeUnit.MILLISECONDS); + + Assertions.assertTrue(latch.await(5, TimeUnit.SECONDS), "Task should have run once"); + // Give a would-be re-run every chance to happen before asserting it didn't. + Thread.sleep(100); + Assertions.assertEquals(1, count.get()); + } + + @Test + @Tag("regressiontest") + void cancellingFutureStopsFurtherRunsTest() throws Exception { + addDescription("Test that cancelling the returned ScheduledFuture stops further dispatches of that task."); + executor = new ScheduledVirtualThreadExecutor("cancelTest", true); + AtomicInteger count = new AtomicInteger(); + CountDownLatch firstRun = new CountDownLatch(1); + + ScheduledFuture future = executor.scheduleAtFixedRate(() -> { + count.incrementAndGet(); + firstRun.countDown(); + }, 0, 20, TimeUnit.MILLISECONDS); + + Assertions.assertTrue(firstRun.await(5, TimeUnit.SECONDS), "Task should have run at least once"); + future.cancel(false); + int countAtCancellation = count.get(); + Thread.sleep(150); + + Assertions.assertEquals(countAtCancellation, count.get(), "No further runs should have happened after cancellation"); + } + + @Test + @Tag("regressiontest") + void taskRunsOnVirtualThreadTest() throws Exception { + addDescription("Test that dispatched tasks run on their own virtual thread, not the ticker thread."); + executor = new ScheduledVirtualThreadExecutor("virtualThreadTest", true); + AtomicBoolean isVirtual = new AtomicBoolean(false); + CountDownLatch latch = new CountDownLatch(1); + + executor.schedule(() -> { + isVirtual.set(Thread.currentThread().isVirtual()); + latch.countDown(); + }, 0, TimeUnit.MILLISECONDS); + + Assertions.assertTrue(latch.await(5, TimeUnit.SECONDS), "Task should have run"); + Assertions.assertTrue(isVirtual.get(), "Dispatched task should run on a virtual thread"); + } + + @Test + @Tag("regressiontest") + void exceptionInTaskDoesNotStopFutureRunsTest() throws Exception { + addDescription("Test that an uncaught exception in a dispatched task doesn't stop subsequent scheduled runs, " + + "unlike java.util.Timer which silently kills the whole timer thread on the first uncaught exception."); + executor = new ScheduledVirtualThreadExecutor("exceptionTest", true); + CountDownLatch latch = new CountDownLatch(3); + + executor.scheduleAtFixedRate(() -> { + latch.countDown(); + throw new RuntimeException("Deliberate failure from test"); + }, 0, 20, TimeUnit.MILLISECONDS); + + Assertions.assertTrue(latch.await(5, TimeUnit.SECONDS), + "Task should keep running despite throwing every time"); + } + + @Test + @Tag("regressiontest") + void exceptionInTaskIsLoggedTest() throws Exception { + addDescription("Test that an uncaught exception in a dispatched task is logged instead of vanishing silently."); + ch.qos.logback.classic.Logger rootLogger = (ch.qos.logback.classic.Logger) LoggerFactory.getLogger( + ch.qos.logback.classic.Logger.ROOT_LOGGER_NAME); + @SuppressWarnings("unchecked") + Appender mockAppender = mock(Appender.class); + when(mockAppender.getName()).thenReturn("MOCK"); + rootLogger.addAppender(mockAppender); + + try (AutoCloseable ignored = () -> rootLogger.detachAppender(mockAppender)) { + executor = new ScheduledVirtualThreadExecutor("loggingTest", true); + String failureMessage = "Deliberate failure for logging test"; + + executor.schedule(() -> { + throw new RuntimeException(failureMessage); + }, 0, TimeUnit.MILLISECONDS); + + ArgumentCaptor argument = ArgumentCaptor.forClass(ILoggingEvent.class); + verify(mockAppender, timeout(5000)).doAppend(argument.capture()); + ILoggingEvent logLine = argument.getValue(); + + Assertions.assertEquals(Level.ERROR, logLine.getLevel()); + Assertions.assertEquals(ScheduledVirtualThreadExecutorTest.class.getName(), logLine.getLoggerName()); + Assertions.assertEquals(failureMessage, logLine.getThrowableProxy().getMessage()); + } + } + + @Test + @Tag("regressiontest") + void closeStopsFurtherDispatchTest() throws Exception { + addDescription("Test that close() stops the ticker so no further dispatches occur."); + executor = new ScheduledVirtualThreadExecutor("closeTest", true); + AtomicInteger count = new AtomicInteger(); + CountDownLatch firstRun = new CountDownLatch(1); + + executor.scheduleAtFixedRate(() -> { + count.incrementAndGet(); + firstRun.countDown(); + }, 0, 20, TimeUnit.MILLISECONDS); + + Assertions.assertTrue(firstRun.await(5, TimeUnit.SECONDS), "Task should have run at least once"); + executor.close(); + int countAtClose = count.get(); + Thread.sleep(150); + + Assertions.assertEquals(countAtClose, count.get(), "No further runs should have happened after close()"); + } +} diff --git a/bitrepository-integrity-service/src/main/java/org/bitrepository/integrityservice/IntegrityServiceManager.java b/bitrepository-integrity-service/src/main/java/org/bitrepository/integrityservice/IntegrityServiceManager.java index ac63f1e9e..7ebc70e3d 100644 --- a/bitrepository-integrity-service/src/main/java/org/bitrepository/integrityservice/IntegrityServiceManager.java +++ b/bitrepository-integrity-service/src/main/java/org/bitrepository/integrityservice/IntegrityServiceManager.java @@ -212,6 +212,9 @@ public void start() { @Override public void shutdown() { + if (workFlowManager != null) { + workFlowManager.shutdown(); + } if (messageBus != null) { try { messageBus.close(); diff --git a/bitrepository-monitoring-service/src/main/java/org/bitrepository/monitoringservice/collector/StatusCollector.java b/bitrepository-monitoring-service/src/main/java/org/bitrepository/monitoringservice/collector/StatusCollector.java index 7d18d8ecf..c1d566a0e 100644 --- a/bitrepository-monitoring-service/src/main/java/org/bitrepository/monitoringservice/collector/StatusCollector.java +++ b/bitrepository-monitoring-service/src/main/java/org/bitrepository/monitoringservice/collector/StatusCollector.java @@ -23,14 +23,14 @@ import org.bitrepository.access.getstatus.GetStatusClient; import org.bitrepository.client.eventhandler.EventHandler; +import org.bitrepository.common.ScheduledVirtualThreadExecutor; import org.bitrepository.common.settings.Settings; import org.bitrepository.common.utils.XmlUtils; import org.bitrepository.monitoringservice.alarm.MonitorAlerter; import org.bitrepository.monitoringservice.status.StatusStore; import java.time.Duration; -import java.util.Timer; -import java.util.TimerTask; +import java.util.concurrent.TimeUnit; /** * The collector of status messages. @@ -41,7 +41,8 @@ public class StatusCollector { private final EventHandler eventHandler; private static final boolean TIMER_IS_DAEMON = true; private static final String NAME_OF_TIMER = "GetStatus collection timer"; - private static final Timer timer = new Timer(NAME_OF_TIMER, TIMER_IS_DAEMON); + private static final ScheduledVirtualThreadExecutor scheduler = + new ScheduledVirtualThreadExecutor(NAME_OF_TIMER, TIMER_IS_DAEMON); /** Collection interval in milliseconds */ private final long collectionInterval; @@ -65,14 +66,14 @@ public StatusCollector(GetStatusClient getStatusClient, Settings settings, Statu * Start the collection of statuses */ public void start() { - timer.schedule(new StatusCollectorTimerTask(), 0, collectionInterval); + scheduler.scheduleWithFixedDelay(new StatusCollectorTimerTask(), 0, collectionInterval, TimeUnit.MILLISECONDS); } /** * Stop the collection of statuses */ public void stop() { - timer.cancel(); + scheduler.close(); } /** @@ -80,7 +81,7 @@ public void stop() { * Tells the store that a new status request has been issued, and then starts the conversation for retrieving the * status from all the contributors. */ - private class StatusCollectorTimerTask extends TimerTask { + private class StatusCollectorTimerTask implements Runnable { @Override public void run() { statusStore.updateReplyCounts(); diff --git a/bitrepository-reference-pillar/src/main/java/org/bitrepository/pillar/Pillar.java b/bitrepository-reference-pillar/src/main/java/org/bitrepository/pillar/Pillar.java index 9c0f50351..98598a7f6 100644 --- a/bitrepository-reference-pillar/src/main/java/org/bitrepository/pillar/Pillar.java +++ b/bitrepository-reference-pillar/src/main/java/org/bitrepository/pillar/Pillar.java @@ -107,6 +107,7 @@ private void initializeWorkflows() { */ public void close() { try { + scheduler.shutdown(); mediator.close(); messageBus.close(); pillarModel.close(); diff --git a/bitrepository-service/src/main/java/org/bitrepository/service/scheduler/JobScheduler.java b/bitrepository-service/src/main/java/org/bitrepository/service/scheduler/JobScheduler.java index 582594219..decf7c5b0 100644 --- a/bitrepository-service/src/main/java/org/bitrepository/service/scheduler/JobScheduler.java +++ b/bitrepository-service/src/main/java/org/bitrepository/service/scheduler/JobScheduler.java @@ -89,4 +89,9 @@ default Instant getNextRunInstant(JobID jobId) { * @param listener The callback listener to receive the events. */ void addJobEventListener(JobEventListener listener); + + /** + * Stops the scheduler and releases its resources. No further jobs will run after this returns. + */ + void shutdown(); } diff --git a/bitrepository-service/src/main/java/org/bitrepository/service/scheduler/TimerBasedScheduler.java b/bitrepository-service/src/main/java/org/bitrepository/service/scheduler/TimerBasedScheduler.java index af866c724..401e77e45 100644 --- a/bitrepository-service/src/main/java/org/bitrepository/service/scheduler/TimerBasedScheduler.java +++ b/bitrepository-service/src/main/java/org/bitrepository/service/scheduler/TimerBasedScheduler.java @@ -24,6 +24,7 @@ */ package org.bitrepository.service.scheduler; +import org.bitrepository.common.ScheduledVirtualThreadExecutor; import org.bitrepository.common.utils.TimeUtils; import org.bitrepository.service.workflow.JobID; import org.bitrepository.service.workflow.JobTimerTask; @@ -39,15 +40,17 @@ import java.util.LinkedList; import java.util.List; import java.util.Map; -import java.util.Timer; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; /** * Scheduler that uses Timer to run workflows. */ public class TimerBasedScheduler implements JobScheduler { private final Logger log = LoggerFactory.getLogger(getClass()); - private final Timer timer; + private final ScheduledVirtualThreadExecutor scheduler; private final Map intervalTasks = new HashMap<>(); + private final Map> scheduledFutures = new HashMap<>(); public static final long SCHEDULE_INTERVAL = 60000; private static final String TIMER_NAME = "Service Scheduler"; private static final boolean TIMER_IS_DAEMON = true; @@ -58,7 +61,7 @@ public class TimerBasedScheduler implements JobScheduler { * Sets up a timer task for running the workflows at requested interval. */ public TimerBasedScheduler() { - timer = new Timer(TIMER_NAME, TIMER_IS_DAEMON); + scheduler = new ScheduledVirtualThreadExecutor(TIMER_NAME, TIMER_IS_DAEMON); } @Override @@ -67,7 +70,7 @@ public void schedule(SchedulableJob workflow, Long interval) { JobTimerTask task = new JobTimerTask(interval, workflow, Collections.unmodifiableList(jobListeners)); if (interval > 0) { - scheduleJob(task); + scheduledFutures.put(workflow.getJobID(), scheduleJob(task)); } intervalTasks.put(workflow.getJobID(), task); @@ -87,7 +90,7 @@ public String startJob(SchedulableJob job) { } JobTimerTask task = new JobTimerTask(timeBetweenRuns, job, Collections.unmodifiableList(jobListeners)); - scheduleJob(task); + scheduledFutures.put(job.getJobID(), scheduleJob(task)); intervalTasks.put(job.getJobID(), task); return "Job scheduled"; } @@ -129,11 +132,19 @@ public JobTimerTask cancelJob(JobID jobID) { if (task == null) { return null; } - task.cancel(); + ScheduledFuture future = scheduledFutures.remove(jobID); + if (future != null) { + future.cancel(false); + } return task; } + @Override + public void shutdown() { + scheduler.close(); + } + /** * Schedules a task. * If the interval for the task is > 0, then it should be scheduled to run at fixed interval, @@ -141,11 +152,11 @@ public JobTimerTask cancelJob(JobID jobID) { * * @param task The task to schedule. */ - private void scheduleJob(JobTimerTask task) { + private ScheduledFuture scheduleJob(JobTimerTask task) { if (task.getIntervalBetweenRuns() > 0) { - timer.scheduleAtFixedRate(task, NO_DELAY, SCHEDULE_INTERVAL); + return scheduler.scheduleAtFixedRate(task, NO_DELAY, SCHEDULE_INTERVAL, TimeUnit.MILLISECONDS); } else { - timer.schedule(task, NO_DELAY); + return scheduler.schedule(task, NO_DELAY, TimeUnit.MILLISECONDS); } } } diff --git a/bitrepository-service/src/main/java/org/bitrepository/service/workflow/JobTimerTask.java b/bitrepository-service/src/main/java/org/bitrepository/service/workflow/JobTimerTask.java index 2903e9998..50803aa14 100644 --- a/bitrepository-service/src/main/java/org/bitrepository/service/workflow/JobTimerTask.java +++ b/bitrepository-service/src/main/java/org/bitrepository/service/workflow/JobTimerTask.java @@ -28,13 +28,12 @@ import java.time.Instant; import java.util.Date; import java.util.List; -import java.util.TimerTask; /** * A timer task encapsulating a workflow. * Used for scheduling workflows to run continuously at a given interval. */ -public class JobTimerTask extends TimerTask { +public class JobTimerTask implements Runnable { private final Logger log = LoggerFactory.getLogger(getClass()); private Instant nextRun; private final long interval; diff --git a/bitrepository-service/src/main/java/org/bitrepository/service/workflow/WorkflowManager.java b/bitrepository-service/src/main/java/org/bitrepository/service/workflow/WorkflowManager.java index c8dae21e3..da108cb73 100644 --- a/bitrepository-service/src/main/java/org/bitrepository/service/workflow/WorkflowManager.java +++ b/bitrepository-service/src/main/java/org/bitrepository/service/workflow/WorkflowManager.java @@ -66,6 +66,13 @@ public String startWorkflow(JobID jobID) { return scheduler.startJob(getWorkflow(jobID)); } + /** + * Stops the underlying scheduler. No further workflow runs will be triggered after this returns. + */ + public void shutdown() { + scheduler.shutdown(); + } + public List getWorkflows(String collectionID) { return collectionWorkflows.get(collectionID); } diff --git a/bitrepository-service/src/test/java/org/bitrepository/service/scheduler/TimerBasedSchedulerTest.java b/bitrepository-service/src/test/java/org/bitrepository/service/scheduler/TimerBasedSchedulerTest.java new file mode 100644 index 000000000..2e1f6550c --- /dev/null +++ b/bitrepository-service/src/test/java/org/bitrepository/service/scheduler/TimerBasedSchedulerTest.java @@ -0,0 +1,222 @@ +/* + * #%L + * Bitrepository Service + * %% + * Copyright (C) 2010 - 2026 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 + * . + * #L% + */ +package org.bitrepository.service.scheduler; + +import org.bitrepository.TestGroups; +import org.bitrepository.service.workflow.JobID; +import org.bitrepository.service.workflow.JobTimerTask; +import org.bitrepository.service.workflow.SchedulableJob; +import org.bitrepository.service.workflow.WorkflowContext; +import org.bitrepository.service.workflow.WorkflowState; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.bitrepository.common.utils.AllureTestUtils.addDescription; + +class TimerBasedSchedulerTest { + private TimerBasedScheduler scheduler; + + @AfterEach + void tearDown() { + if (scheduler != null) { + scheduler.shutdown(); + } + } + + @Test + @Tag(TestGroups.REGRESSIONTEST) + void scheduleWithPositiveIntervalRunsJobPromptlyTest() throws Exception { + addDescription("Test that schedule() with a positive interval dispatches the job right away."); + scheduler = new TimerBasedScheduler(); + FakeJob job = new FakeJob(new JobID("workflow", "collection")); + + scheduler.schedule(job, 1000L); + + Assertions.assertTrue(job.started.await(5, TimeUnit.SECONDS), "Job should have been started"); + } + + @Test + @Tag(TestGroups.REGRESSIONTEST) + void scheduleWithNonPositiveIntervalRegistersButDoesNotRunTheJobTest() throws Exception { + addDescription("Test that schedule() with a non-positive interval registers the job without triggering a run, " + + "per the interval==never contract documented on JobScheduler#schedule."); + scheduler = new TimerBasedScheduler(); + FakeJob job = new FakeJob(new JobID("workflow", "collection")); + + scheduler.schedule(job, 0L); + + Assertions.assertFalse(job.started.await(200, TimeUnit.MILLISECONDS), "Job should not have been started"); + Assertions.assertEquals(0, scheduler.getRunInterval(job.getJobID()), "The job should still be registered"); + Assertions.assertNull(scheduler.getNextRunInstant(job.getJobID()), + "A job with a non-positive interval has no next run"); + } + + @Test + @Tag(TestGroups.REGRESSIONTEST) + void cancelJobReturnsScheduledTaskAndStopsItTest() throws Exception { + addDescription("Test that cancelJob() returns the scheduled task and removes it from the scheduler's bookkeeping."); + scheduler = new TimerBasedScheduler(); + FakeJob job = new FakeJob(new JobID("workflow", "collection")); + scheduler.schedule(job, 1000L); + Assertions.assertTrue(job.started.await(5, TimeUnit.SECONDS), "Job should have been started"); + + JobTimerTask cancelled = scheduler.cancelJob(job.getJobID()); + + Assertions.assertNotNull(cancelled); + Assertions.assertEquals(job.getJobID(), cancelled.getWorkflowID()); + Assertions.assertEquals(-1, scheduler.getRunInterval(job.getJobID()), + "A cancelled job is no longer known to the scheduler"); + Assertions.assertNull(scheduler.cancelJob(job.getJobID()), "Cancelling an unknown job returns null"); + } + + @Test + @Tag(TestGroups.REGRESSIONTEST) + void startJobStartsAnIdleJobImmediatelyTest() throws Exception { + addDescription("Test that startJob() runs a job that isn't currently running right away."); + scheduler = new TimerBasedScheduler(); + FakeJob job = new FakeJob(new JobID("workflow", "collection")); + + String result = scheduler.startJob(job); + + Assertions.assertEquals("Job scheduled", result); + Assertions.assertTrue(job.started.await(5, TimeUnit.SECONDS), "Job should have been started"); + } + + @Test + @Tag(TestGroups.REGRESSIONTEST) + void startJobRefusesAJobThatIsAlreadyRunningTest() { + addDescription("Test that startJob() doesn't run a job that's already in a non-idle state."); + scheduler = new TimerBasedScheduler(); + FakeJob job = new FakeJob(new JobID("workflow", "collection")); + job.setCurrentState(WorkflowState.RUNNING); + + String result = scheduler.startJob(job); + + Assertions.assertEquals("Already running", result); + Assertions.assertEquals(0, job.startCount.get(), "The already-running job should not have been started again"); + } + + @Test + @Tag(TestGroups.REGRESSIONTEST) + void jobEventListenerIsNotifiedWhenJobFinishesTest() throws Exception { + addDescription("Test that a registered JobEventListener is notified once the dispatched job finishes."); + scheduler = new TimerBasedScheduler(); + FakeJob job = new FakeJob(new JobID("workflow", "collection")); + CountDownLatch notified = new CountDownLatch(1); + scheduler.addJobEventListener(new JobEventListener() { + @Override + public void jobStarted(SchedulableJob j) { + } + + @Override + public void jobFailed(SchedulableJob j) { + } + + @Override + public void jobFinished(SchedulableJob j) { + if (j.getJobID().equals(job.getJobID())) { + notified.countDown(); + } + } + }); + + scheduler.schedule(job, 1000L); + + Assertions.assertTrue(notified.await(5, TimeUnit.SECONDS), "Listener should have been notified"); + } + + @Test + @Tag(TestGroups.REGRESSIONTEST) + void getRunIntervalAndNextRunAreUnknownForAnUnscheduledJobTest() { + addDescription("Test that querying a job that was never scheduled returns the documented \"unknown\" values."); + scheduler = new TimerBasedScheduler(); + JobID unknownJob = new JobID("neverScheduled", "collection"); + + Assertions.assertEquals(-1, scheduler.getRunInterval(unknownJob)); + Assertions.assertNull(scheduler.getNextRunInstant(unknownJob)); + } + + @Test + @Tag(TestGroups.REGRESSIONTEST) + void shutdownStopsTheSchedulerTest() { + addDescription("Test that shutdown() releases the scheduler's resources without throwing."); + scheduler = new TimerBasedScheduler(); + FakeJob job = new FakeJob(new JobID("workflow", "collection")); + scheduler.schedule(job, 1000L); + + Assertions.assertDoesNotThrow(() -> scheduler.shutdown()); + } + + private static class FakeJob implements SchedulableJob { + private final JobID id; + private final CountDownLatch started = new CountDownLatch(1); + private final AtomicInteger startCount = new AtomicInteger(); + private volatile WorkflowState state = WorkflowState.NOT_RUNNING; + + FakeJob(JobID id) { + this.id = id; + } + + @Override + public void start() { + startCount.incrementAndGet(); + // Simulate a job that finishes synchronously, so repeated scheduling isn't blocked by its state. + state = WorkflowState.NOT_RUNNING; + started.countDown(); + } + + @Override + public WorkflowState currentState() { + return state; + } + + @Override + public void setCurrentState(WorkflowState newState) { + state = newState; + } + + @Override + public String getHumanReadableState() { + return state.toString(); + } + + @Override + public String getDescription() { + return "FakeJob for " + id; + } + + @Override + public JobID getJobID() { + return id; + } + + @Override + public void initialise(WorkflowContext context, String collectionID) { + } + } +}