Shut down a directory poller's thread pool when the poller stops

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

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<String, Long> trackedFiles = new HashMap<String, Long>(); 
@@ -101,6 +104,50 @@ public void init(Session session, Map<String, String> 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<String> failures) {
         try {

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.
+ * <p>
+ * 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));
+    }
+}

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