diff --git a/RELEASE-NOTES.md b/RELEASE-NOTES.md index 6a56a991..814f61c9 100644 --- a/RELEASE-NOTES.md +++ b/RELEASE-NOTES.md @@ -1,27 +1,19 @@ # OpenAS2 Server -# Version 4.6.0 +# Version 4.6.2 # RELEASE NOTES ----- -The OpenAS2 project is pleased to announce the release of OpenAS2 4.6.0 +The OpenAS2 project is pleased to announce the release of OpenAS2 4.6.2 -The release download file is: OpenAS2Server-4.6.0.zip +The release download file is: OpenAS2Server-4.6.2.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.0 - 2025-08-04 - -This is a testing enhancement release. - -1. Reworked the resource access class -2. changed tests to use the same config that is provided in the installer package -3. Optimised method to wait for files to appear on file system as part of end-to-end tests -4. Created script to generate test certificates keystore with updated certificates -5. Added maven phase to regenerate test certificates keystore prior to running tests -6. Updated Maven wrapper for project -7. Added Java 24 to test matrix +Version 4.6.2 - 2025-09-27 +This is a bugfix release. +1. Enhanced the poller algorithm to avoid race conditions under very high volume file processing. ##Upgrade Notes @@ -31,6 +23,9 @@ This is a testing enhancement release. **You must review all notes for the relevant intermediate versions from your version to this release version.** +### Upgrading to 4.6.1 or newer from any older version if using parallel processing mode: + 1. Ensure you change the property for enabling parallel mode by removing the 3rd consecutive "l" from "process_files_in_paralllel". + ### Upgrading to 4.0 or newer from any older version: 1. Ensure you implement all logging that you had configured for earlier versions using the logback configuration or replace with another framework that works with SLF4J facade. See the OpenAS2HowTo.pdf logging section for more details. 2. The property for email configuration in the config.xml changed: diff --git a/Server/pom.xml b/Server/pom.xml index 86febd97..e9bec853 100644 --- a/Server/pom.xml +++ b/Server/pom.xml @@ -7,7 +7,7 @@ net.sf.openas2 OpenAS2 - 4.6.0 + 4.6.2 ../pom.xml diff --git a/Server/src/bin/import_public_cert.bat b/Server/src/bin/import_public_cert.bat index dec26246..cf1a4e38 100755 --- a/Server/src/bin/import_public_cert.bat +++ b/Server/src/bin/import_public_cert.bat @@ -73,7 +73,7 @@ if "%action%" == "replace" ( if errorlevel 1 ( echo. echo ***** Failed to import the certificate to the keystore. See errors above to correct the problem. - echo If the error shows the certifcate already eists then add the "replace" option to the command line. + echo If the error shows the certifcate already exists then add the "replace" option to the command line. EXIT /B 1 ) diff --git a/Server/src/bin/start-openas2.bat b/Server/src/bin/start-openas2.bat index ee11887b..e1dfb398 100755 --- a/Server/src/bin/start-openas2.bat +++ b/Server/src/bin/start-openas2.bat @@ -24,15 +24,15 @@ set OPENAS2_CONFIG_FILE=%OPENAS2_BASE_DIR%/config/config.xml :skip_config_file_set for %%F in ("%OPENAS2_CONFIG_FILE%") do set OPENAS2_CONFIG_DIR=%%~dpF -if [%OPENAS2_PROPERTIES_FILE%]==[]] goto skip_properties_file +if [%OPENAS2_PROPERTIES_FILE%]==[] goto skip_properties_file set EXTRA_PARMS=%EXTRA_PARMS% -Dopenas2.properties.file="%OPENAS2_PROPERTIES_FILE%" :skip_properties_file rem set EXTRA_PARMS=%EXTRA_PARMS% -Dhttps.protocols=TLSv1.2 -if NOT [%OPENAS2_LOGGING_BASE%]==[]] goto skip_logging_base_set +if NOT [%OPENAS2_LOGGING_BASE%]==[] goto skip_logging_base_set set OPENAS2_LOGGING_BASE=%OPENAS2_BASE_DIR%\logs :skip_logging_base_set -set EXTRA_PARMS=%EXTRA_PARMS% -DOPENAS2_LOGGING_BASE=OPENAS2_LOGGING_BASE +set EXTRA_PARMS=%EXTRA_PARMS% DOPENAS2_LOG_DIR="%OPENAS2_LOGGING_BASE%" rem Uncomment any of the following for enhanced debug rem set EXTRA_PARMS=%EXTRA_PARMS% -Dmaillogger.debug.enabled=true diff --git a/Server/src/config/as2_certs.p12 b/Server/src/config/as2_certs.p12 index 47d0232a..779333d8 100644 Binary files a/Server/src/config/as2_certs.p12 and b/Server/src/config/as2_certs.p12 differ diff --git a/Server/src/config/config.xml b/Server/src/config/config.xml index a242e8d6..0a1eea90 100644 --- a/Server/src/config/config.xml +++ b/Server/src/config/config.xml @@ -81,8 +81,9 @@ pollerConfigBase.defaults="sender.as2_id=$partnership.sender.as2_id$, receiver.as2_id=$partnership.receiver.as2_id$" pollerConfigBase.sendfilename="true" pollerConfigBase.mimetype="application/EDI-X12" - pollerConfigBase.process_files_in_paralllel="false" + pollerConfigBase.process_files_in_parallel="false" pollerConfigBase.max_parallel_files="20" + pollerConfigBase.max_file_processing_time_minutes="30" partnerships.polling.interval="120" messages.enabled="false" messages.polling.interval="120" @@ -206,8 +207,9 @@ defaults="$properties.pollerConfigBase.defaults$" sendfilename="$properties.pollerConfigBase.sendfilename$" mimetype="$properties.pollerConfigBase.mimetype$" - process_files_in_paralllel="$properties.pollerConfigBase.process_files_in_paralllel$" - max_parallel_files="$properties.pollerConfigBase.max_parallel_files$"/> + process_files_in_parallel="$properties.pollerConfigBase.process_files_in_parallel$" + max_parallel_files="$properties.pollerConfigBase.max_parallel_files$" + max_file_processing_time_minutes="$properties.pollerConfigBase.max_file_processing_time_minutes$" /> diff --git a/Server/src/config/openas2.properties.sample b/Server/src/config/openas2.properties.sample index dea476f3..25b3ed14 100644 --- a/Server/src/config/openas2.properties.sample +++ b/Server/src/config/openas2.properties.sample @@ -107,7 +107,7 @@ pollerConfigBase.interval=5 pollerConfigBase.defaults=sender.as2_id=$partnership.sender.as2_id$, receiver.as2_id=$partnership.receiver.as2_id$ pollerConfigBase.sendfilename=true pollerConfigBase.mimetype=application/EDI-X12 -pollerConfigBase.process_files_in_paralllel=false +pollerConfigBase.process_files_in_parallel=false pollerConfigBase.max_parallel_files=20 # The time between checks for a changed partnerships.xml file for auto reload. -partnerships.polling.interval=120 \ No newline at end of file +partnerships.polling.interval=120 diff --git a/Server/src/main/java/CheckCertificate.java b/Server/src/main/java/CheckCertificate.java index ccd543f6..eee840be 100644 --- a/Server/src/main/java/CheckCertificate.java +++ b/Server/src/main/java/CheckCertificate.java @@ -1,7 +1,7 @@ import org.apache.commons.cli.CommandLine; import org.apache.commons.cli.CommandLineParser; import org.apache.commons.cli.DefaultParser; -import org.apache.commons.cli.HelpFormatter; +import org.apache.commons.cli.help.HelpFormatter; import org.apache.commons.cli.Option; import org.apache.commons.cli.Options; import org.apache.commons.cli.ParseException; @@ -64,8 +64,13 @@ private void usage(Options options) { String header = "Checks SSL connectivity." + "\nTries to connect to the remote server and establish a connection."; String footer = "Good luck!"; - HelpFormatter formatter = new HelpFormatter(); - formatter.printHelp(this.getClass().getName(), header, options, footer, true); + HelpFormatter formatter = HelpFormatter.builder().get(); + try { + formatter.printHelp(this.getClass().getName(), header, options, footer, true); + } catch (IOException e) { + // TODO Auto-generated catch block + e.printStackTrace(); + } } private CommandLine parseCommandLine(String[] args) { @@ -75,7 +80,7 @@ private CommandLine parseCommandLine(String[] args) { // create the Options Options options = new Options(); for (String[] opt : opts) { - Option option = Option.builder(opt[0]).longOpt(opt[1]).hasArg("true".equalsIgnoreCase(opt[2])).desc(opt[4]).build(); + Option option = Option.builder(opt[0]).longOpt(opt[1]).hasArg("true".equalsIgnoreCase(opt[2])).desc(opt[4]).get(); option.setRequired("true".equalsIgnoreCase(opt[3])); options.addOption(option); } diff --git a/Server/src/main/java/SplitCsvFile.java b/Server/src/main/java/SplitCsvFile.java index 271d5b26..635c23df 100644 --- a/Server/src/main/java/SplitCsvFile.java +++ b/Server/src/main/java/SplitCsvFile.java @@ -1,12 +1,13 @@ - import org.apache.commons.cli.CommandLine; - import org.apache.commons.cli.CommandLineParser; - import org.apache.commons.cli.DefaultParser; - import org.apache.commons.cli.HelpFormatter; - import org.apache.commons.cli.Option; - import org.apache.commons.cli.Options; - import org.apache.commons.cli.ParseException; - import org.openas2.util.FileUtil; - import java.io.File; +import org.apache.commons.cli.CommandLine; +import org.apache.commons.cli.CommandLineParser; +import org.apache.commons.cli.DefaultParser; +import org.apache.commons.cli.help.HelpFormatter; +import org.apache.commons.cli.Option; +import org.apache.commons.cli.Options; +import org.apache.commons.cli.ParseException; +import org.openas2.util.FileUtil; +import java.io.File; +import java.io.IOException; /** * Class used to add the server's certificate to the KeyStore with your trusted @@ -38,8 +39,13 @@ private void usage(Options options) { String header = "Splits CSV file." + "\nReads the file as a line based file creating new files that will not exceed the specified maximum size."; String footer = "NOTE: The file is expected to contain lines separated by newline characters."; - HelpFormatter formatter = new HelpFormatter(); - formatter.printHelp(this.getClass().getName(), header, options, footer, true); + HelpFormatter formatter = HelpFormatter.builder().get(); + try { + formatter.printHelp(this.getClass().getName(), header, options, footer, true); + } catch (IOException e) { + // TODO Auto-generated catch block + e.printStackTrace(); + } } private CommandLine parseCommandLine(String[] args) { @@ -49,7 +55,7 @@ private CommandLine parseCommandLine(String[] args) { // create the Options Options options = new Options(); for (String[] opt : opts) { - Option option = Option.builder(opt[0]).longOpt(opt[1]).hasArg("true".equalsIgnoreCase(opt[2])).desc(opt[4]).build(); + Option option = Option.builder(opt[0]).longOpt(opt[1]).hasArg("true".equalsIgnoreCase(opt[2])).desc(opt[4]).get(); option.setRequired("true".equalsIgnoreCase(opt[3])); options.addOption(option); } diff --git a/Server/src/main/java/org/openas2/app/cert/ImportCertInEncodedStreamCommand.java b/Server/src/main/java/org/openas2/app/cert/ImportCertInEncodedStreamCommand.java index 16806fda..24f67c86 100644 --- a/Server/src/main/java/org/openas2/app/cert/ImportCertInEncodedStreamCommand.java +++ b/Server/src/main/java/org/openas2/app/cert/ImportCertInEncodedStreamCommand.java @@ -8,9 +8,7 @@ import java.io.ByteArrayInputStream; import java.io.IOException; -import java.security.cert.Certificate; import java.security.cert.CertificateException; -import java.security.cert.X509Certificate; public class ImportCertInEncodedStreamCommand extends AliasedCertCommand { public String getDefaultDescription() { @@ -22,7 +20,7 @@ public String getDefaultName() { } public String getDefaultUsage() { - return "importbybstream "; + return "importbystream "; } public CommandResult execute(AliasedCertificateFactory certFx, Object[] params) throws OpenAS2Exception { diff --git a/Server/src/main/java/org/openas2/cmd/processor/BaseCommandProcessor.java b/Server/src/main/java/org/openas2/cmd/processor/BaseCommandProcessor.java index 6add1430..59f1cce4 100644 --- a/Server/src/main/java/org/openas2/cmd/processor/BaseCommandProcessor.java +++ b/Server/src/main/java/org/openas2/cmd/processor/BaseCommandProcessor.java @@ -14,7 +14,6 @@ import java.util.Map; import java.util.concurrent.Callable; import java.util.concurrent.ScheduledExecutorService; -import org.slf4j.Logger; import org.slf4j.LoggerFactory; diff --git a/Server/src/main/java/org/openas2/params/CompositeParameters.java b/Server/src/main/java/org/openas2/params/CompositeParameters.java index 006bcebe..fccb0efc 100644 --- a/Server/src/main/java/org/openas2/params/CompositeParameters.java +++ b/Server/src/main/java/org/openas2/params/CompositeParameters.java @@ -34,8 +34,8 @@ public CompositeParameters add(String key, ParameterParser param) { public void setParameter(String key, String value) throws InvalidParameterException { StringTokenizer keyParts = new StringTokenizer(key, ".", false); - keyParts.nextToken(); - ParameterParser parser = getParameterParsers().get(keyParts); + String parserKey = keyParts.nextToken(); + ParameterParser parser = getParameterParsers().get(parserKey); if (parser != null) { if (!keyParts.hasMoreTokens()) { diff --git a/Server/src/main/java/org/openas2/processor/msgtracking/DbTrackingModule.java b/Server/src/main/java/org/openas2/processor/msgtracking/DbTrackingModule.java index d6b97fa3..52c7e85d 100644 --- a/Server/src/main/java/org/openas2/processor/msgtracking/DbTrackingModule.java +++ b/Server/src/main/java/org/openas2/processor/msgtracking/DbTrackingModule.java @@ -17,7 +17,6 @@ import java.sql.PreparedStatement; import java.sql.ResultSet; import java.sql.ResultSetMetaData; -import java.sql.SQLException; import java.sql.Statement; import java.sql.Types; import java.util.ArrayList; 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 03b2059a..c388ca71 100644 --- a/Server/src/main/java/org/openas2/processor/receiver/DirectoryPollingModule.java +++ b/Server/src/main/java/org/openas2/processor/receiver/DirectoryPollingModule.java @@ -19,8 +19,6 @@ import java.util.*; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; public abstract class DirectoryPollingModule extends PollingModule { @@ -30,18 +28,27 @@ public abstract class DirectoryPollingModule extends PollingModule { public static final String PARAM_FILE_EXTENSION_FILTER = "fileextensionfilter"; public static final String PARAM_FILE_EXTENSION_EXCLUDE_FILTER = "fileextensionexcludefilter"; public static final String PARAM_FILE_NAME_EXCLUDE_FILTER = "filenameexcluderegexfilter"; - public static final String PARAM_PROCESS_IN_PARALLEL = "process_files_in_paralllel"; + public static final String PARAM_PROCESS_IN_PARALLEL = "process_files_in_parallel"; public static final String PARAM_MAX_PARALLEL_FILES = "max_parallel_files"; - private Map trackedFiles; + 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; + + // 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(); + // Files that have been passed to a processor - key is the absolute file path and value is the processing start timestamp + private Map processingFiles = new HashMap(); + private String errorDir = null; private String sentDir = null; private boolean processFilesAsThreads = false; private int maxProcessingThreads = 20; + private long maxFileProcessingTime = MAX_FILE_PROCESSING_TIME_DEFAULT_MINUTES*60*1000; // support fixed size thread group to run file processing as threads private ExecutorService executorService = null; - private List allowExtensions; - private List excludeExtensions; - private String excludeFilenameRegexFilter = null; + private List allowExtensions = new ArrayList(); + private List excludeExtensions = new ArrayList(); + private String excludeFilenameRegexFilter = ""; private Logger logger = LoggerFactory.getLogger(DirectoryPollingModule.class); @@ -69,23 +76,21 @@ public void init(Session session, Map options) throws OpenAS2Exc IOUtil.getDirectoryFile(pendingInfoFolder); String pendingFolder = getSession().getProcessor().getParameters().get(Processor.PENDING_MDN_MSG_DIRECTORY_IDENTIFIER); IOUtil.getDirectoryFile(pendingFolder); - - + String maxFileProcessingTimeFromPAram = getParameter(PARAM_MAX_FILE_PROCESSING_TIME_MINUTES, ""); + if (maxFileProcessingTimeFromPAram.length() > 0) { + maxFileProcessingTime = Long.parseLong(maxFileProcessingTimeFromPAram) * 60 * 1000; + } String allowExtensionFilter = getParameter(PARAM_FILE_EXTENSION_FILTER, ""); String excludeExtensionFilter = getParameter(PARAM_FILE_EXTENSION_EXCLUDE_FILTER, ""); String excludeFilenameRegexFilter = getParameter(PARAM_FILE_NAME_EXCLUDE_FILTER, ""); - if (allowExtensionFilter == null || allowExtensionFilter.length() < 1) { - this.allowExtensions = new ArrayList(); - } else { + if (allowExtensionFilter.length() > 0) { this.allowExtensions = Arrays.asList(allowExtensionFilter.split("\\s*,\\s*")); } - if (excludeExtensionFilter == null || excludeExtensionFilter.length() < 1) { - this.excludeExtensions = new ArrayList(); - } else { + if (excludeExtensionFilter.length() > 0) { this.excludeExtensions = Arrays.asList(excludeExtensionFilter.split("\\s*,\\s*")); } - if (excludeFilenameRegexFilter != null && excludeFilenameRegexFilter.length() > 0) { + if (excludeFilenameRegexFilter.length() > 0) { this.excludeFilenameRegexFilter = excludeFilenameRegexFilter; } @@ -125,11 +130,8 @@ public void poll() { protected void scanDirectory(String directory) throws IOException, InvalidParameterException { - /* Claudio.Degioanni - Versione modificata 20210628 2.11 - Start */ - /** - * Rispetto alla versione tesi ho aggiunto il supporto per allowExtensions e - * excludeExtensions aggiunto nelle versioni dopo + * Support allowed extensions excluded extensions. */ if (logger.isTraceEnabled()) { logger.trace("Polling module scanning directory: " + directory); @@ -137,50 +139,49 @@ protected void scanDirectory(String directory) throws IOException, InvalidParame File directoryAsFile = IOUtil.getDirectoryFile(directory); // Wrap in try-with-resources block to ensure close() is called try (DirectoryStream dirs = Files.newDirectoryStream(directoryAsFile.toPath(), entry -> { - String name = entry.getFileName().toString(); if (Files.isDirectory(entry)) { return false; } + if (isTracked(entry.toAbsolutePath().toString())) { + return false; + } + String name = entry.getFileName().toString(); if (logger.isTraceEnabled()) { logger.trace("Polling module file name found: " + name); } String extension = name.substring(name.lastIndexOf(".") + 1); - boolean isAllowed = true; - if (!allowExtensions.isEmpty()) { - isAllowed = allowExtensions.contains(extension); + if (!allowExtensions.isEmpty() && !allowExtensions.contains(extension)) { + // There is a list of allowed extensions and this one not in it so do not include + return false; } - // Check for the excluded filters if not already disallowed - if (isAllowed && !excludeExtensions.isEmpty()) { - isAllowed = !excludeExtensions.contains(extension); + if (!excludeExtensions.isEmpty() && excludeExtensions.contains(extension)) { + // This is a disallowed extension so ignore this file + return false; } // Check if there are filename regex exclusions if not already disallowed - if (isAllowed && excludeFilenameRegexFilter != null) { - isAllowed = !name.matches(excludeFilenameRegexFilter); + if (excludeFilenameRegexFilter.length() > 0 && name.matches(excludeFilenameRegexFilter)) { + // This is a disallowed extension so ignore this file + return false; } - return isAllowed; - + return true; })) { - - for (Path dir : dirs) { - File currentFile = dir.toFile(); - - if (checkFile(currentFile)) { - // start watching the file's size if it's not already being watched - trackFile(currentFile); + for (Path dir : dirs) { + File currentFile = dir.toFile(); + // Don't track locked files as this can create false positives when file is being created + if (!fileIsLocked(currentFile)) { + // Found an untracked file so add it + trackedFiles.put(currentFile.getAbsolutePath(), currentFile.length()); + } } - } - /* Claudio.degioanni - Versione modificata 20210628 2.11 - End */ } } - protected boolean checkFile(File file) { - if (file.exists() && file.isFile()) { + protected boolean fileIsLocked(File file) { + if (file.isFile()) { try { - // check for a write-lock on file, will skip file if it's write - // locked + // check for a write-lock on file, will skip file if it's write locked FileOutputStream fOut = new FileOutputStream(file, true); fOut.close(); - return true; } catch (IOException ioe) { // a sharing violation occurred, ignore the file for now if (logger.isDebugEnabled()) { @@ -190,17 +191,18 @@ protected boolean checkFile(File file) { e.printStackTrace(); } } + return true; } } return false; } - private void trackFile(File file) { - Map trackedFiles = getTrackedFiles(); - String filePath = file.getAbsolutePath(); - if (trackedFiles.get(filePath) == null) { - trackedFiles.put(filePath, file.length()); - } + private boolean isTracked(String filePath) { + return trackedFiles.containsKey(filePath); + } + + private boolean isBeingProcessed(String filePath) { + return processingFiles.containsKey(filePath); } protected void processSingleFile(File file, String fileEntryKey) { @@ -216,53 +218,70 @@ protected void processSingleFile(File file, String fileEntryKey) { return; } } finally { + // Remove trackedFiles entry first to avoid race condition in parallel processing trackedFiles.remove(fileEntryKey); + processingFiles.remove(fileEntryKey); } } - private void updateTracking() { - // clone the trackedFiles map, iterator through the clone and modify the - // original to avoid iterator exceptions - // is there a better way to do this? - Map trackedFiles = getTrackedFiles(); - Map trackedFilesClone = new HashMap(trackedFiles); + 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); + } + } - for (Map.Entry fileEntry : trackedFilesClone.entrySet()) { + private void updateTracking() { + // Use an iterator to be able to remove entries whilst iterating over the map. + Iterator> iter = trackedFiles.entrySet().iterator(); + while (iter.hasNext()) { + Map.Entry fileEntry = iter.next(); // get the file and it's stored length String fileEntryKey = fileEntry.getKey(); - File file = new File(fileEntry.getKey()); - long fileLength = fileEntry.getValue().longValue(); + File file = new File(fileEntryKey); + long trackedFileLength = fileEntry.getValue().longValue(); // if the file no longer exists, remove it from the tracker - if (!checkFile(file)) { - trackedFiles.remove(fileEntryKey); + if (!file.exists()) { + iter.remove(); + processingFiles.remove(fileEntryKey); + } else if (isBeingProcessed(fileEntryKey)) { + // Handle failed processing of files based on a max timeout + long elapsedTime = System.currentTimeMillis() - processingFiles.get(fileEntryKey); + if (elapsedTime > maxFileProcessingTime) { + iter.remove(); + processingFiles.remove(fileEntryKey); + logger.error("WARNING: A file that was being processed has exceeded the processing time limit and has been removed from the processing tracker: " + fileEntryKey + + "\n\tThis will trigger a retry to send the file. If this causes duplicate sends then try to increase the " + PARAM_MAX_FILE_PROCESSING_TIME_MINUTES + " value."); + } } else { - // if the file length has changed, update the tracker + // if the file length has increased, update the tracker long newLength = file.length(); - if (newLength != fileLength) { + if (newLength > trackedFileLength) { trackedFiles.put(fileEntryKey, Long.valueOf(newLength)); } else { - // if the file length has stayed the same, process the file - // and then stop tracking it - if (processFilesAsThreads) { - if (logger.isDebugEnabled()) { - logger.debug("Parallel processing mode handling file: " + file.getName()); - } - executorService.execute(new Runnable() { - @Override - public void run() { - processSingleFile(file, fileEntryKey); - } - }); - - } else { - processSingleFile(file, fileEntryKey); - } + // no change in file length so process the file + triggerFileProcessing(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() { + processSingleFile(file, fileEntryKey); + } + }); + } + protected abstract Message createMessage(); protected void processFile(File file) throws OpenAS2Exception { @@ -279,11 +298,4 @@ protected void processFile(File file) throws OpenAS2Exception { throw new OpenAS2Exception("Failed to initiate processing for file:" + file.getAbsolutePath(), e); } } - - private Map getTrackedFiles() { - if (trackedFiles == null) { - trackedFiles = new HashMap(); - } - return trackedFiles; - } -} + } diff --git a/Server/src/main/java/org/openas2/processor/resender/DirectoryResenderModule.java b/Server/src/main/java/org/openas2/processor/resender/DirectoryResenderModule.java index 525eea74..8a2734c9 100644 --- a/Server/src/main/java/org/openas2/processor/resender/DirectoryResenderModule.java +++ b/Server/src/main/java/org/openas2/processor/resender/DirectoryResenderModule.java @@ -26,8 +26,6 @@ import java.util.List; import java.util.Map; import java.util.StringTokenizer; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; public class DirectoryResenderModule extends BaseResenderModule { diff --git a/Server/src/main/java/org/openas2/schedule/SchedulerComponent.java b/Server/src/main/java/org/openas2/schedule/SchedulerComponent.java index 56bc43f8..8e4cf7bc 100644 --- a/Server/src/main/java/org/openas2/schedule/SchedulerComponent.java +++ b/Server/src/main/java/org/openas2/schedule/SchedulerComponent.java @@ -38,7 +38,7 @@ public void init(Session session, Map parameters) throws OpenAS2 private void createExecutor() throws InvalidParameterException { int configuredAmountOfThreads = getParameterInt(PARAMETER_THREADS, false); int amountOfThreads = configuredAmountOfThreads < MIN_AMOUNT_OF_THREADS ? MIN_AMOUNT_OF_THREADS : configuredAmountOfThreads; - BasicThreadFactory threadFactory = new BasicThreadFactory.Builder().namingPattern(getName() + "-Thread-%d").build(); + BasicThreadFactory threadFactory = BasicThreadFactory.builder().namingPattern(getName() + "-Thread-%d").build(); this.executorService = Executors.newScheduledThreadPool(amountOfThreads, threadFactory); logger.debug("Scheduler module is ready."); diff --git a/Server/src/test/java/org/openas2/TestPartner.java b/Server/src/test/java/org/openas2/TestPartner.java index 034a68c1..fd347c86 100644 --- a/Server/src/test/java/org/openas2/TestPartner.java +++ b/Server/src/test/java/org/openas2/TestPartner.java @@ -70,6 +70,10 @@ public String getName() { return name; } + public Partnership getPartnership() { + return partnership; + } + public String getPartnerName() { return partnerName; } diff --git a/Server/src/test/java/org/openas2/TestUtils.java b/Server/src/test/java/org/openas2/TestUtils.java index 7526622b..5e00c435 100644 --- a/Server/src/test/java/org/openas2/TestUtils.java +++ b/Server/src/test/java/org/openas2/TestUtils.java @@ -3,8 +3,19 @@ import java.io.File; import java.io.FileNotFoundException; import java.io.FilenameFilter; +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.StandardOpenOption; +import java.util.Iterator; +import java.util.List; import java.util.concurrent.TimeUnit; +import org.openas2.app.OpenAS2Server; +import org.openas2.partner.Partnership; +import org.openas2.partner.PartnershipFactory; +import org.openas2.processor.receiver.DirectoryPollingModule; + /** * Utilities for tests */ @@ -45,4 +56,69 @@ public static boolean deleteDirectory(File directoryToBeDeleted) { } return directoryToBeDeleted.delete(); } + + public static int waitTillAllFilesReceived(File dir, int fileCount, int timeout, TimeUnit unit) throws FileNotFoundException { + long finishAt = System.currentTimeMillis() + unit.toMillis(timeout); + while (finishAt - System.currentTimeMillis() > 0) { + File[] files = dir.listFiles(); + if (files.length == fileCount) { + return fileCount; + } else if (files.length > fileCount) { + return files.length; + } + } + return dir.listFiles().length; + } + + /** + * Finds the first partnership in the list for the server instance that has a directory poller and builds a TestPartner object using that + * @param server - the instance of an OpenAS2 server + * @return - a TestPartner instance based on the partnership found + * @throws Exception + */ + public static TestPartner getFromFirstSendingPartnership(OpenAS2Server server) throws Exception { + PartnershipFactory pf = server.getSession().getPartnershipFactory(); + List partnerships = pf.getPartnerships(); + for (Iterator iterator = partnerships.iterator(); iterator.hasNext();) { + Partnership partnership = (Partnership) iterator.next(); + DirectoryPollingModule pollerModule = getPollingModule((XMLSession) server.getSession(), partnership); + if (pollerModule != null) { + return new TestPartner(server, partnership, pollerModule); + } + } + return null; + } + + public static TestPartner getFromPartnerIds(OpenAS2Server server, String senderAs2Id, String receiverAs2Id) throws Exception { + PartnershipFactory pf = server.getSession().getPartnershipFactory(); + List partnerships = pf.getPartnerships(); + for (Iterator iterator = partnerships.iterator(); iterator.hasNext();) { + Partnership partnership = (Partnership) iterator.next(); + if (senderAs2Id.equals(partnership.getSenderID(Partnership.PID_AS2)) && receiverAs2Id.equals(partnership.getReceiverID(Partnership.PID_AS2))) { + DirectoryPollingModule pollerModule = getPollingModule((XMLSession) server.getSession(), partnership); + return new TestPartner(server, partnership, pollerModule); + } + } + return null; + } + + public static DirectoryPollingModule getPollingModule(XMLSession session, Partnership partnership) throws ComponentNotFoundException { + DirectoryPollingModule dirPollMod = session.getPartnershipPoller(partnership.getName()); + if (dirPollMod != null) { + return dirPollMod; + } + // Try to find a module defined poller since there is no matching poller by name. (config.xml defined pollers do not have the correct partnership name in the poller cache) + return session.getPartnershipPoller(partnership.getSenderID(Partnership.PID_AS2), partnership.getReceiverID(Partnership.PID_AS2)); + } + + public static void appendLinesToFile(Path filePath, String[] lines) throws IOException { + for (int i = 0; i < lines.length; i++) { + Files.write( + filePath, + ("\n" + lines[i]).getBytes(), + StandardOpenOption.APPEND); + } + + } + } diff --git a/Server/src/test/java/org/openas2/app/OpenAS2ServerTest.java b/Server/src/test/java/org/openas2/app/OpenAS2ServerTest.java index 8331ef07..8f21daaa 100644 --- a/Server/src/test/java/org/openas2/app/OpenAS2ServerTest.java +++ b/Server/src/test/java/org/openas2/app/OpenAS2ServerTest.java @@ -1,20 +1,15 @@ package org.openas2.app; import org.apache.commons.io.FileUtils; -import org.apache.commons.io.filefilter.PrefixFileFilter; import org.apache.commons.lang3.RandomStringUtils; import org.apache.commons.lang3.exception.ExceptionUtils; import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; -import org.openas2.ComponentNotFoundException; import org.openas2.TestPartner; import org.openas2.TestResource; -import org.openas2.XMLSession; -import org.openas2.partner.Partnership; -import org.openas2.partner.PartnershipFactory; -import org.openas2.processor.receiver.DirectoryPollingModule; +import org.openas2.TestUtils; import org.openas2.util.Properties; import java.io.File; @@ -23,7 +18,6 @@ import java.nio.file.Files; import java.nio.file.Paths; import java.util.ArrayList; -import java.util.Iterator; import java.util.List; import java.util.concurrent.Callable; import java.util.concurrent.ExecutorService; @@ -64,19 +58,19 @@ public static void startServers() throws Exception { // Get the data folder from Properties before starting the other server as it overwrites the Properties dataFolders[0] = Properties.getProperty("storageBaseDir", null); // Iterate over the partnerships picking the first one that has a directory poller - server1PartnerSender = getFromFirstSendingPartnership(server1); + server1PartnerSender = TestUtils.getFromFirstSendingPartnership(server1); // Use the 2 partners in the initial partnership to get other parnerships to test both way transfer // Get a receiver partnership for the matching partners in the sender partnership for server A - server1PartnerReceiver = getFromPartnerIds(server1, server1PartnerSender.getPartnerAS2Id(), server1PartnerSender.getAs2Id()); + server1PartnerReceiver = TestUtils.getFromPartnerIds(server1, server1PartnerSender.getPartnerAS2Id(), server1PartnerSender.getAs2Id()); System.setProperty("SERVER2_PARTNERSHIP_FILE", TestResource.getResource("server2-partnerships")); String props2File = TestResource.getResource("server2-props"); System.setProperty(Properties.OPENAS2_PROPERTIES_FILE_PROP, props2File); server2 = new OpenAS2Server.Builder().run(configFile); // Set up the receiver fin server B for the sender from server A - server2PartnerReceiver = getFromPartnerIds(server2, server1PartnerSender.getAs2Id(), server1PartnerSender.getPartnerAS2Id()); + server2PartnerReceiver = TestUtils.getFromPartnerIds(server2, server1PartnerSender.getAs2Id(), server1PartnerSender.getPartnerAS2Id()); // Get a sender partnership for the matching partners in the receiver partnership for server B - server2PartnerSender = getFromPartnerIds(server2, server1PartnerSender.getPartnerAS2Id(), server1PartnerSender.getAs2Id()); + server2PartnerSender = TestUtils.getFromPartnerIds(server2, server1PartnerSender.getPartnerAS2Id(), server1PartnerSender.getAs2Id()); dataFolders[1] = Properties.getProperty("storageBaseDir", null); executorService = Executors.newFixedThreadPool(20); } catch (FileNotFoundException e) { @@ -181,56 +175,6 @@ private void verifyMessageDelivery(TestMessage testMessage) throws IOException { assertThat("Verify sent MDN was stored by " + testMessage.fromPartner.getName(), sentMDN.exists(), is(true)); } - /** - * Finds the first partnership in the list for the server instance that has a directory poller and builds a TestPartner object using that - * @param server - the instance of an OpenAS2 server - * @return - a TestPartner instance based on the partnership found - * @throws Exception - */ - private static TestPartner getFromFirstSendingPartnership(OpenAS2Server server) throws Exception { - PartnershipFactory pf = server.getSession().getPartnershipFactory(); - List partnerships = pf.getPartnerships(); - for (Iterator iterator = partnerships.iterator(); iterator.hasNext();) { - Partnership partnership = (Partnership) iterator.next(); - DirectoryPollingModule pollerModule = getPollingModule((XMLSession) server.getSession(), partnership); - if (pollerModule != null) { - return new TestPartner(server, partnership, pollerModule); - } - } - return null; - } - - private static TestPartner getFromPartnerIds(OpenAS2Server server, String senderAs2Id, String receiverAs2Id) throws Exception { - PartnershipFactory pf = server.getSession().getPartnershipFactory(); - List partnerships = pf.getPartnerships(); - for (Iterator iterator = partnerships.iterator(); iterator.hasNext();) { - Partnership partnership = (Partnership) iterator.next(); - if (senderAs2Id.equals(partnership.getSenderID(Partnership.PID_AS2)) && receiverAs2Id.equals(partnership.getReceiverID(Partnership.PID_AS2))) { - DirectoryPollingModule pollerModule = getPollingModule((XMLSession) server.getSession(), partnership); - return new TestPartner(server, partnership, pollerModule); - } - } - return null; - } - - private static DirectoryPollingModule getPollingModule(XMLSession session, Partnership partnership) throws ComponentNotFoundException { - DirectoryPollingModule dirPollMod = session.getPartnershipPoller(partnership.getName()); - if (dirPollMod != null) { - return dirPollMod; - } - // Try to find a module defined poller since there is no matching poller by name. (config.xml defined pollers do not have the correct partnership name in the poller cache) - return session.getPartnershipPoller(partnership.getSenderID(Partnership.PID_AS2), partnership.getReceiverID(Partnership.PID_AS2)); - } - - @SuppressWarnings("unused") - private void setPartnershipToAsync(Partnership partnership) throws Exception { - if (partnership != null) { - partnership.setAttribute(Partnership.PA_AS2_RECEIPT_OPTION, "http://localhost:20081"); - } else { - throw new Exception("Could not set partnership to ASYNC mode"); - } - } - private static class TestMessage { private final String fileName; private final String body; diff --git a/Server/src/test/java/org/openas2/app/ParallelProcessingTest.java b/Server/src/test/java/org/openas2/app/ParallelProcessingTest.java new file mode 100644 index 00000000..a14f0285 --- /dev/null +++ b/Server/src/test/java/org/openas2/app/ParallelProcessingTest.java @@ -0,0 +1,148 @@ +package org.openas2.app; + +import org.apache.commons.io.FileUtils; +import org.apache.commons.lang3.RandomStringUtils; +import org.apache.commons.lang3.exception.ExceptionUtils; +import org.hamcrest.Matchers; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import org.openas2.TestPartner; +import org.openas2.TestResource; +import org.openas2.TestUtils; +import org.openas2.XMLSession; +import org.openas2.processor.ActiveModule; +import org.openas2.processor.ProcessorModule; +import org.openas2.processor.receiver.AS2DirectoryPollingModule; +import org.openas2.processor.receiver.DirectoryPollingModule; +import org.openas2.processor.storage.MessageFileModule; +import org.openas2.processor.storage.StorageModule; +import org.openas2.util.Properties; + +import java.io.File; +import java.io.FileNotFoundException; +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.util.List; +import java.util.concurrent.TimeUnit; + +import static org.hamcrest.MatcherAssert.assertThat; +import static org.openas2.TestUtils.waitTillAllFilesReceived; + + +public class ParallelProcessingTest { + + + private static TestPartner server1PartnerSender; + private static TestPartner server2PartnerReceiver; + private static OpenAS2Server server1; + private static OpenAS2Server server2; + private static String[] dataFolders = new String[2]; + private final int msgCnt = 1000; + + @TempDir + public static File resourceTmp; + @TempDir + public static File configTmp; + + @BeforeAll + public static void startServers() throws Exception { + resourceTmp = Files.createTempDirectory("testResources").toFile(); + //System.setProperty("OPENAS2_LOG_LEVEL", "TRACE"); + try { + String configFile = TestResource.getResource("config"); + System.setProperty("SERVER1_PARTNERSHIP_FILE", TestResource.getResource("server1-partnerships")); + String propsSrcFile = TestResource.getResource("server1-props"); + String props1File = configTmp.getAbsolutePath() + File.separator + "parallel_test.properties"; + Path propsPath = Paths.get(props1File); + Files.copy(Paths.get(propsSrcFile), propsPath); + String[] newProps = {"pollerConfigBase.process_files_in_parallel=true"}; + TestUtils.appendLinesToFile(propsPath, newProps); + System.setProperty(Properties.OPENAS2_PROPERTIES_FILE_PROP, props1File); + server1 = new OpenAS2Server.Builder().run(configFile); + // Get the data folder from Properties before starting the other server as it overwrites the Properties + dataFolders[0] = Properties.getProperty("storageBaseDir", null); + // Iterate over the partnerships picking the first one that has a directory poller + server1PartnerSender = TestUtils.getFromFirstSendingPartnership(server1); + + System.setProperty("SERVER2_PARTNERSHIP_FILE", TestResource.getResource("server2-partnerships")); + String props2File = TestResource.getResource("server2-props"); + System.setProperty(Properties.OPENAS2_PROPERTIES_FILE_PROP, props2File); + server2 = new OpenAS2Server.Builder().run(configFile); + // Set up the receiver server B to receive files from server A + server2PartnerReceiver = TestUtils.getFromPartnerIds(server2, server1PartnerSender.getAs2Id(), server1PartnerSender.getPartnerAS2Id()); + dataFolders[1] = Properties.getProperty("storageBaseDir", null); + } catch (FileNotFoundException e) { + System.err.println("Failed to retrieve resource for test: " + e.getMessage()); + e.printStackTrace(); + throw new Exception(e); + } catch (Throwable e) { + // TODO Auto-generated catch block + e.printStackTrace(); + throw new Exception(e); + } + } + + @Test + public void shouldSendMessagesInParallel() throws Exception { + try { + createMessages(msgCnt); + sendAllMessages(server1PartnerSender, server2PartnerReceiver); + int rxdCount = getDeliveredMessagesCount(msgCnt); + assertThat("Received number of files", rxdCount, Matchers.equalTo(msgCnt)); + } catch (Throwable e) { + // Aid debugging JUnit test failures + System.out.println("shouldSendMessagesSyncMdn ERROR OCCURRED: " + ExceptionUtils.getStackTrace(e)); + throw new Exception(e); + } + } + + @Test + public void shouldBeParallelMode() throws Exception { + XMLSession session = (XMLSession)server1.getSession(); + DirectoryPollingModule pollingModule = session.getPartnershipPoller(server1PartnerSender.getPartnership().getName()); + String modeSetting = pollingModule.getParameter(DirectoryPollingModule.PARAM_PROCESS_IN_PARALLEL, "false"); + assertThat("Directory polling module running in parallel mode", "true", Matchers.equalTo(modeSetting)); + } + + @AfterAll + public static void tearDown() throws Exception { + server1PartnerSender.getServer().shutdown(); + server2PartnerReceiver.getServer().shutdown(); + // Cleanup the folders created so the test does not fail next time round from leftover data + // NOTE: For debugging "missing" files it is best to comment this out + for (int i = 0; i < dataFolders.length; i++) { + try { + FileUtils.deleteDirectory(new File(dataFolders[i])); + } catch (IOException e) { + // TODO Auto-generated catch block + e.printStackTrace(); + } + } + } + + private void createMessages(int tgtCount) throws IOException { + for (int i = 0; i < tgtCount; i++) { + String outgoingMsgFileName = RandomStringUtils.secure().nextAlphanumeric(10) + ".txt"; + String outgoingMsgBody = RandomStringUtils.secure().nextAlphanumeric(1024); + File msgfile = Files.createFile(Paths.get(resourceTmp.toString(), outgoingMsgFileName)).toFile(); + FileUtils.write(msgfile, outgoingMsgBody, "UTF-8"); + } + } + + private void sendAllMessages(TestPartner fromPartner, TestPartner toPartner) throws IOException { + File[] files = resourceTmp.listFiles(); + System.out.println("Moving " + files.length + " files to outbox for sending to:" + fromPartner.getOutbox()); + for(int i = 0; i < files.length; i++) { + FileUtils.moveFileToDirectory(files[i], fromPartner.getOutbox(), false); + } + } + + private int getDeliveredMessagesCount(int expectedCnt) throws IOException { + return waitTillAllFilesReceived(server2PartnerReceiver.getInbox(), expectedCnt, 45, TimeUnit.SECONDS); + } + +} diff --git a/Server/src/test/java/org/openas2/support/config/FileMonitorAdapterTest.java b/Server/src/test/java/org/openas2/support/config/FileMonitorAdapterTest.java index 170a97ab..9e186ce4 100644 --- a/Server/src/test/java/org/openas2/support/config/FileMonitorAdapterTest.java +++ b/Server/src/test/java/org/openas2/support/config/FileMonitorAdapterTest.java @@ -71,7 +71,7 @@ public void shouldNotScheduleRefreshWhenIntervalNotConfigured() throws Exception @Test public void shouldScheduleConfigRefresh() throws Exception { - int refreshInterval = RandomUtils.nextInt(1, 10); + int refreshInterval = RandomUtils.secure().randomInt(1, 10); doReturn(true).when(configFile).exists(); doReturn(true).when(configFile).isFile(); diff --git a/Server/src/test/java/org/openas2/util/IOUtilTest.java b/Server/src/test/java/org/openas2/util/IOUtilTest.java index d128c8a4..1db4de13 100644 --- a/Server/src/test/java/org/openas2/util/IOUtilTest.java +++ b/Server/src/test/java/org/openas2/util/IOUtilTest.java @@ -47,7 +47,6 @@ public Record(String filename, String format, String delimiters, boolean mergeEx } } - @SuppressWarnings("serial") private List testParameterParsingData = new ArrayList() { { add(new Record("MyCo-PartnerCo-MyFileNameWithExtension.edi", "sender.as2_id, receiver.as2_id," + " attributes.filename", "-.", false, "MyFileNameWithExtension", "MyCo", "PartnerCo")); diff --git a/changes.txt b/changes.txt index 118b6e5c..14fa40d7 100644 --- a/changes.txt +++ b/changes.txt @@ -1,5 +1,18 @@ **IMPORTANT NOTE**: Please review upgrade notes in the RELEASE-NOTES.md if you are upgrading +Version 4.6.2 - 2025-09-27 + +This is a bugfix release. +1. Enhanced the poller algorithm to avoid race conditions under very high volume file processing. + +Version 4.6.1 - 2025-09-25 + +This is a bugfix release. + +1. Enhanced the poller algorithm to avoid problems with high volume file processing. +2. Fixed the attribute name to enable the parallel processing mode (paralllel - > parallel) + + Version 4.6.0 - 2025-08-04 This is a testing enhancement release. diff --git a/docs/OpenAS2HowTo.odt b/docs/OpenAS2HowTo.odt index 6f4f32bb..5571192e 100644 Binary files a/docs/OpenAS2HowTo.odt and b/docs/OpenAS2HowTo.odt differ diff --git a/docs/OpenAS2HowTo.pdf b/docs/OpenAS2HowTo.pdf index 993efd5c..5bae5243 100644 Binary files a/docs/OpenAS2HowTo.pdf and b/docs/OpenAS2HowTo.pdf differ diff --git a/pom.xml b/pom.xml index 85cc5298..0ead45c5 100644 --- a/pom.xml +++ b/pom.xml @@ -5,7 +5,7 @@ 4.0.0 net.sf.openas2 OpenAS2 - 4.6.0 + 4.6.2 OpenAS2 pom @@ -51,17 +51,17 @@ org.bouncycastle bcjmail-jdk18on - 1.81 + 1.82 org.bouncycastle bcpkix-jdk18on - 1.81 + 1.82 org.bouncycastle bcprov-jdk18on - 1.81 + 1.82 org.bouncycastle @@ -71,7 +71,7 @@ org.bouncycastle bcpg-jdk18on - 1.81 + 1.82 org.apache.commons @@ -116,14 +116,14 @@ org.mockito mockito-core - 5.18.0 + 5.20.0 test org.mockito mockito-junit-jupiter - 5.18.0 + 5.20.0 test @@ -161,50 +161,50 @@ org.glassfish.jersey.containers jersey-container-grizzly2-http - 3.1.10 + 3.1.11 jar com.fasterxml.jackson.core jackson-databind - 2.19.2 + 2.20.0 jar com.fasterxml.jackson.module jackson-module-jaxb-annotations - 2.19.2 + 2.20.0 org.glassfish.jersey.media jersey-media-json-jackson - 3.1.10 + 3.1.11 jar org.glassfish.jersey.inject jersey-hk2 - 3.1.10 + 3.1.11 jakarta.xml.bind jakarta.xml.bind-api - 4.0.2 + 4.0.4 com.sun.xml.bind jaxb-core - 4.0.5 + 4.0.6 com.sun.xml.bind jaxb-impl - 4.0.5 + 4.0.6 com.zaxxer HikariCP - 7.0.0 + 7.0.2