From 89ee86143119c41de36f57aa742bf196a005a3cb Mon Sep 17 00:00:00 2001 From: kaah Date: Tue, 14 Jul 2026 15:53:57 +0200 Subject: [PATCH 1/2] Replace java.util.Timer with ScheduledExecutorService + virtual threads An uncaught exception in any TimerTask silently kills the whole timer thread. ScheduledExecutorService is the recommended replacement, and combined with virtual threads it lets each firing run independently without needing a bounded thread pool. --- .../AuditTrailCollectionTimerTask.java | 3 +- .../collector/AuditTrailCollector.java | 15 +- .../preserver/LocalAuditTrailPreserver.java | 24 +- .../CollectionBasedConversationMediator.java | 18 +- .../ScheduledVirtualThreadExecutor.java | 101 +++++++++ .../ScheduledVirtualThreadExecutorTest.java | 206 ++++++++++++++++++ .../IntegrityServiceManager.java | 3 + .../collector/StatusCollector.java | 13 +- .../java/org/bitrepository/pillar/Pillar.java | 1 + .../service/scheduler/JobScheduler.java | 5 + .../scheduler/TimerBasedScheduler.java | 29 ++- .../service/workflow/JobTimerTask.java | 3 +- .../service/workflow/WorkflowManager.java | 7 + 13 files changed, 383 insertions(+), 45 deletions(-) create mode 100644 bitrepository-core/src/main/java/org/bitrepository/common/ScheduledVirtualThreadExecutor.java create mode 100644 bitrepository-core/src/test/java/org/bitrepository/common/ScheduledVirtualThreadExecutorTest.java 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..9dbf9eaad --- /dev/null +++ b/bitrepository-core/src/main/java/org/bitrepository/common/ScheduledVirtualThreadExecutor.java @@ -0,0 +1,101 @@ +/* + * #%L + * Bitmagasin integrationstest + * + * $Id$ + * $HeadURL$ + * %% + * Copyright (C) 2010 The State and University Library, 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..9f4da6f4d --- /dev/null +++ b/bitrepository-core/src/test/java/org/bitrepository/common/ScheduledVirtualThreadExecutorTest.java @@ -0,0 +1,206 @@ +/* + * #%L + * Bitmagasin integrationstest + * + * $Id$ + * $HeadURL$ + * %% + * Copyright (C) 2010 The State and University Library, 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); } From 3f9cc1d7ccaa42bca68ee59e5a0c0f8b9873d8e8 Mon Sep 17 00:00:00 2001 From: kaah Date: Mon, 24 Aug 2026 14:42:44 +0200 Subject: [PATCH 2/2] Added test for TimerBasedScheduler functionality --- .../ScheduledVirtualThreadExecutor.java | 4 +- .../ScheduledVirtualThreadExecutorTest.java | 4 +- .../scheduler/TimerBasedSchedulerTest.java | 222 ++++++++++++++++++ 3 files changed, 226 insertions(+), 4 deletions(-) create mode 100644 bitrepository-service/src/test/java/org/bitrepository/service/scheduler/TimerBasedSchedulerTest.java diff --git a/bitrepository-core/src/main/java/org/bitrepository/common/ScheduledVirtualThreadExecutor.java b/bitrepository-core/src/main/java/org/bitrepository/common/ScheduledVirtualThreadExecutor.java index 9dbf9eaad..89d22582f 100644 --- a/bitrepository-core/src/main/java/org/bitrepository/common/ScheduledVirtualThreadExecutor.java +++ b/bitrepository-core/src/main/java/org/bitrepository/common/ScheduledVirtualThreadExecutor.java @@ -1,11 +1,11 @@ /* * #%L - * Bitmagasin integrationstest + * Bitrepository Protocol * * $Id$ * $HeadURL$ * %% - * Copyright (C) 2010 The State and University Library, The Royal Library and The State Archives, Denmark + * 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 diff --git a/bitrepository-core/src/test/java/org/bitrepository/common/ScheduledVirtualThreadExecutorTest.java b/bitrepository-core/src/test/java/org/bitrepository/common/ScheduledVirtualThreadExecutorTest.java index 9f4da6f4d..b57b46780 100644 --- a/bitrepository-core/src/test/java/org/bitrepository/common/ScheduledVirtualThreadExecutorTest.java +++ b/bitrepository-core/src/test/java/org/bitrepository/common/ScheduledVirtualThreadExecutorTest.java @@ -1,11 +1,11 @@ /* * #%L - * Bitmagasin integrationstest + * Bitrepository Protocol * * $Id$ * $HeadURL$ * %% - * Copyright (C) 2010 The State and University Library, The Royal Library and The State Archives, Denmark + * 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 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) { + } + } +}