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 ===========================