Fix concurrent processing error (#479)

* Handle concurrent access issues.

* Version bump.

* Update release notes

Author: uhurusurfa

RELEASE-NOTES.md

@@ -1,19 +1,19 @@
 #              OpenAS2 Server
-#              Version 4.6.2
+#              Version 4.6.3
 #              RELEASE NOTES
 -----
-The OpenAS2 project is pleased to announce the release of OpenAS2 4.6.2
+The OpenAS2 project is pleased to announce the release of OpenAS2 4.6.3
 
-The release download file is: OpenAS2Server-4.6.2.zip
+The release download file is: OpenAS2Server-4.6.3.zip
 
 The zip file contains a PDF document (OpenAS2HowTo.pdf) providing information on installing and using the application.
 ## NOTE: Testing covers Java 11 to 21.
 ##       Java 8 is NO LONGER SUPPORTED.
 
-Version 4.6.2 - 2025-09-27
+Version 4.6.3 - 2025-09-29
 
 This is a bugfix release.
-1. Enhanced the poller algorithm to avoid race conditions under very high volume file processing.
+1. Avoid  concurrent processing error that occurs after files after sent as part of the cleanup process.
 
 
 ##Upgrade Notes

Server/pom.xml

@@ -7,7 +7,7 @@
         <!-- DO NOT CHANGE THIS "groupId" WITHOUT CHANGING XMLSession.getManifestAttributes.MANIFEST_VENDOR_ID_ATTRIB -->
         <groupId>net.sf.openas2</groupId>
         <artifactId>OpenAS2</artifactId>
-        <version>4.6.2</version>
+        <version>4.6.3</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
 

Server/src/main/java/org/openas2/processor/receiver/DirectoryPollingModule.java

@@ -38,6 +38,8 @@ public abstract class DirectoryPollingModule extends PollingModule {
     private Map<String, Long> trackedFiles = new HashMap<String, Long>(); 
     // Files that have been passed to a processor - key is the absolute file path and value is the processing start timestamp
     private Map<String, Long> processingFiles = new HashMap<String, Long>(); 
+    // Files that have been processed in thread mode and need removing from the tracked files maps by the main thread
+    Collection<String> threadProcessedFiles = Collections.synchronizedCollection(new ArrayList<String>());
 
     private String errorDir = null;
     private String sentDir = null;
@@ -217,24 +219,37 @@ protected void processSingleFile(File file, String fileEntryKey) {
                 //forceStop(e1);
                 return;
             }
-        } finally {
-            // Remove trackedFiles entry first to avoid race condition in parallel processing
-            trackedFiles.remove(fileEntryKey);
-            processingFiles.remove(fileEntryKey);
-        }        
+        }     
     }
 
-    protected void triggerFileProcessing(File file, String fileEntryKey) {
-        // Add the in processing flag on the file now otherwise in threaded mode there is a race condition
-        processingFiles.put(fileEntryKey, System.currentTimeMillis());
-        if (processFilesAsThreads) {
-            processFileInThread(file, fileEntryKey);
-        } else {
-            processSingleFile(file, fileEntryKey);
+    private void processFileInThread(File file, String fileEntryKey) {
+        if (logger.isDebugEnabled()) {
+            logger.debug("Parallel processing mode handling file: " + file.getName());
         }
+        executorService.execute(new Runnable() {
+            @Override
+            public void run() {
+                try {
+                    processSingleFile(file, fileEntryKey);
+                } finally {
+                    // Add to list for removal from tracking maps when updateTracking is called
+                    threadProcessedFiles.add(fileEntryKey);
+                }
+            }
+        });
     }
 
     private void updateTracking() {
+        // first remove processed files if we are running in parallel processing mode to avoid concurrency issues
+        if (processFilesAsThreads) {
+            synchronized (threadProcessedFiles) {
+                threadProcessedFiles.forEach((fileEntryKey) -> {
+                    trackedFiles.remove(fileEntryKey);
+                    processingFiles.remove(fileEntryKey);
+                });
+                threadProcessedFiles.clear();
+            }
+        }
         // Use an iterator to be able to remove entries whilst iterating over the map.
         Iterator<Map.Entry<String, Long>> iter = trackedFiles.entrySet().iterator();
         while (iter.hasNext()) {
@@ -264,24 +279,24 @@ private void updateTracking() {
                     trackedFiles.put(fileEntryKey, Long.valueOf(newLength));
                 } else {
                     // no change in file length so process the file
-                    triggerFileProcessing(file, fileEntryKey);
+                    // Add the processing flag on the file now otherwise in threaded mode there is a race condition
+                    processingFiles.put(fileEntryKey, System.currentTimeMillis());
+                    if (processFilesAsThreads) {
+                        processFileInThread(file, fileEntryKey);
+                    } else {
+                        try {
+                            processSingleFile(file, fileEntryKey);
+                        } finally {
+                            iter.remove();
+                            processingFiles.remove(fileEntryKey);
+                        }
+                    }
+
                 }
             }
         }
     }
 
-    private void processFileInThread(File file, String fileEntryKey) {
-        if (logger.isDebugEnabled()) {
-            logger.debug("Parallel processing mode handling file: " + file.getName());
-        }
-        executorService.execute(new Runnable() {
-            @Override
-            public void run() {
-                processSingleFile(file, fileEntryKey);
-            }
-        });
-    }
-
     protected abstract Message createMessage();
 
     protected void processFile(File file) throws OpenAS2Exception {

changes.txt

@@ -1,5 +1,10 @@
 **IMPORTANT NOTE**: Please review upgrade notes in the RELEASE-NOTES.md if you are upgrading
 
+Version 4.6.3 - 2025-09-29
+
+This is a bugfix release.
+1. Avoid  concurrent processing error that occurs after files after sent as part of the cleanup process.
+
 Version 4.6.2 - 2025-09-27
 
 This is a bugfix release.

pom.xml

@@ -5,7 +5,7 @@
     <modelVersion>4.0.0</modelVersion>
     <groupId>net.sf.openas2</groupId>
     <artifactId>OpenAS2</artifactId>
-    <version>4.6.2</version>
+    <version>4.6.3</version>
     <name>OpenAS2</name>
     <packaging>pom</packaging>