From 885e7ae66ef217d4d6de42d6ca8e07bf3623186a Mon Sep 17 00:00:00 2001 From: Jake Huneau Date: Thu, 17 Sep 2026 12:02:26 +0000 Subject: [PATCH] Shut down a directory poller's thread pool when the poller stops A directory poller running in parallel mode creates a fixed thread pool in init and nothing ever shut it down. Stopping a poller only cancelled its timer, and destroy is never called on one. Partnership pollers are destroyed and rebuilt every time the partnerships are reloaded, which happens whenever the partnerships file changes. The module object was dropped but its pool's threads stayed alive, and live threads are garbage collection roots, so the pool and everything it referenced survived for the life of the process. Every reload leaked another one. Stopping a poller now shuts its pool down. The shutdown refuses new work but lets a file that is part way through being sent finish, because interrupting a transmission would leave the partner with an incomplete message and this side unsure whether it arrived. If that work has not drained within ten seconds the wait gives up and says so rather than blocking the reload, and the threads end when the work completes. Starting a poller again now gives it a usable pool. Without that, the restart would leave it holding the pool that had just been shut down under it, and a shut down pool rejects everything handed to it, so the poller would have looked alive while failing every file it picked up. Only deployments with "process_files_in_parallel" enabled were affected, since no pool is created otherwise. --- RELEASE-NOTES.md | 6 ++ .../receiver/DirectoryPollingModule.java | 47 +++++++++ .../DirectoryPollingModuleShutdownTest.java | 97 +++++++++++++++++++ changes.txt | 6 ++ 4 files changed, 156 insertions(+) create mode 100644 Server/src/test/java/org/openas2/processor/receiver/DirectoryPollingModuleShutdownTest.java diff --git a/RELEASE-NOTES.md b/RELEASE-NOTES.md index aca53a16..f72f6d35 100644 --- a/RELEASE-NOTES.md +++ b/RELEASE-NOTES.md @@ -24,6 +24,12 @@ This is a bugfix release. reverted by the next refresh. The endpoint now stores the partnerships after a successful update, as it already did after an add or a delete. The database partnership store was unaffected because it persists each change in its own transaction. +3. Fix a directory poller running in parallel mode leaving its thread pool behind when it is stopped. Partnership pollers + are destroyed and rebuilt whenever the partnerships are reloaded, and a pool whose threads are still alive keeps itself + and everything it references from being collected, so a reload leaked one pool per poller for the life of the process. + Stopping a poller now shuts its pool down, letting a file that is part way through being sent finish first, and starting + a poller again gives it a usable pool. Only "process_files_in_parallel" deployments were affected as no pool is created + otherwise. Version 4.12.0 2026-09-09 =========================== diff --git a/Server/src/main/java/org/openas2/processor/receiver/DirectoryPollingModule.java b/Server/src/main/java/org/openas2/processor/receiver/DirectoryPollingModule.java index d1339979..0ca61627 100644 --- a/Server/src/main/java/org/openas2/processor/receiver/DirectoryPollingModule.java +++ b/Server/src/main/java/org/openas2/processor/receiver/DirectoryPollingModule.java @@ -19,6 +19,7 @@ import java.util.*; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; public abstract class DirectoryPollingModule extends PollingModule { @@ -33,6 +34,8 @@ public abstract class DirectoryPollingModule extends PollingModule { public static final String PARAM_MAX_FILE_PROCESSING_TIME_MINUTES = "max_file_processing_time_minutes"; private final int MAX_FILE_PROCESSING_TIME_DEFAULT_MINUTES = 30; + /** How long stopping waits for files already being sent before leaving them to finish on their own. */ + private static final int SHUTDOWN_WAIT_SECONDS = 10; // Files that have been registered by the poller - key is the absolute file path and value is the file size private Map trackedFiles = new HashMap(); @@ -101,6 +104,50 @@ public void init(Session session, Map options) throws OpenAS2Exc } } + @Override + public void doStart() throws OpenAS2Exception { + /* + * A poller that was stopped and started again needs a usable pool: the one it had was shut down + * by doStop and a shut down pool rejects everything handed to it. + */ + if (processFilesAsThreads && (executorService == null || executorService.isShutdown())) { + executorService = Executors.newFixedThreadPool(maxProcessingThreads); + } + super.doStart(); + } + + @Override + public void doStop() throws OpenAS2Exception { + super.doStop(); + /* + * The pool has to be shut down with the poller that owns it. Partnership pollers are destroyed + * and rebuilt whenever the partnerships are reloaded, and a pool whose threads are still alive + * is reachable from those threads, so leaving it behind leaked the pool and everything it holds + * on every reload. + */ + if (executorService != null) { + /* + * Refuse new work but let a file that is part way through being sent finish: interrupting a + * transmission would leave the partner with an incomplete message and this side unsure + * whether it arrived. The threads end once that work drains. + */ + executorService.shutdown(); + try { + if (!executorService.awaitTermination(SHUTDOWN_WAIT_SECONDS, TimeUnit.SECONDS)) { + logger.warn("Directory poller for " + getOutboxDir() + " still had files in progress after " + + SHUTDOWN_WAIT_SECONDS + " seconds. Its threads will end when that work completes."); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + } + + /** Exposed so a test can check the pool is not left running when the poller stops. */ + ExecutorService getExecutorService() { + return executorService; + } + @Override public boolean healthcheck(List failures) { try { diff --git a/Server/src/test/java/org/openas2/processor/receiver/DirectoryPollingModuleShutdownTest.java b/Server/src/test/java/org/openas2/processor/receiver/DirectoryPollingModuleShutdownTest.java new file mode 100644 index 00000000..30aac3bb --- /dev/null +++ b/Server/src/test/java/org/openas2/processor/receiver/DirectoryPollingModuleShutdownTest.java @@ -0,0 +1,97 @@ +package org.openas2.processor.receiver; + +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.io.FileOutputStream; +import java.util.concurrent.ExecutorService; + +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.TestInstance; +import org.junit.jupiter.api.TestInstance.Lifecycle; +import org.openas2.app.BaseServerSetup; +import org.openas2.partner.Partnership; + +/** + * Verifies that a directory poller running in parallel mode does not leave its thread pool behind when + * it is stopped. + *

+ * Partnership pollers are destroyed and rebuilt every time the partnerships are reloaded, which happens + * whenever the partnerships file changes. A pool whose threads are still alive is reachable from those + * threads, so a pool left running kept itself and everything it referenced alive for the life of the + * process, once per reload. Only parallel mode creates a pool, so only parallel mode was affected. + */ +@TestInstance(Lifecycle.PER_CLASS) +public class DirectoryPollingModuleShutdownTest extends BaseServerSetup { + + private DirectoryPollingModule poller; + + @BeforeAll + public void setUp() throws Exception { + super.createFileSystemResources(this.getClass().getName()); + // Only parallel mode creates a pool, so the leak only exists with this turned on + try (FileOutputStream fos = new FileOutputStream(openAS2PropertiesFile)) { + fos.write("pollerConfigBase.process_files_in_parallel=true\n".getBytes()); + } + super.setStartActiveModules(true); + super.setup(); + poller = session.getPartnershipPoller(simpleTestMsg.getPartnership().getName()); + assertNotNull(poller, "the shipped partnerships should give this test a directory poller"); + } + + @AfterAll + public void tearDown() throws Exception { + super.tearDown(); + } + + @Test + public void aRunningPollerHasAUsablePool() { + ExecutorService pool = poller.getExecutorService(); + + assertNotNull(pool, "parallel mode should have created a pool"); + assertFalse(pool.isShutdown(), "a running poller needs a pool that accepts work"); + } + + @Test + public void stoppingThePollerShutsItsPoolDown() throws Exception { + ExecutorService pool = poller.getExecutorService(); + assertFalse(pool.isShutdown()); + + poller.stop(); + + assertTrue(pool.isShutdown(), + "the pool must be shut down with the poller, or every partnerships reload leaks one"); + // Put it back so the ordering of the other tests does not matter + poller.start(); + } + + @Test + public void aPollerThatIsStartedAgainGetsAWorkingPool() throws Exception { + poller.stop(); + ExecutorService shutDownPool = poller.getExecutorService(); + assertTrue(shutDownPool.isShutdown()); + + poller.start(); + + ExecutorService restarted = poller.getExecutorService(); + assertNotNull(restarted); + assertFalse(restarted.isShutdown(), + "a restarted poller must not be left holding the pool that was shut down under it," + + " or it would reject every file it picks up"); + } + + @Test + public void theOutboxIsStillPolledAfterARestart() throws Exception { + // Guards against the restart leaving the poller in a state where it looks alive but cannot work + poller.stop(); + poller.start(); + + assertTrue(poller.isRunning()); + assertNotNull(poller.getOutboxDir()); + assertFalse(poller.getExecutorService().isShutdown()); + assertNotNull(simpleTestMsg.getPartnership().getReceiverID(Partnership.PID_AS2)); + } +} diff --git a/changes.txt b/changes.txt index f21cad06..187f6214 100644 --- a/changes.txt +++ b/changes.txt @@ -15,6 +15,12 @@ This is a bugfix release. reverted by the next refresh. The endpoint now stores the partnerships after a successful update, as it already did after an add or a delete. The database partnership store was unaffected because it persists each change in its own transaction. +3. Fix a directory poller running in parallel mode leaving its thread pool behind when it is stopped. Partnership pollers + are destroyed and rebuilt whenever the partnerships are reloaded, and a pool whose threads are still alive keeps itself + and everything it references from being collected, so a reload leaked one pool per poller for the life of the process. + Stopping a poller now shuts its pool down, letting a file that is part way through being sent finish first, and starting + a poller again gives it a usable pool. Only "process_files_in_parallel" deployments were affected as no pool is created + otherwise. Version 4.12.0 2026-09-09 ===========================