Release abandoned VFS lock files based on their file system age

modules/transports/core/vfs/src/main/java/org/apache/synapse/transport/vfs/PollTableEntry.java

@@ -72,6 +72,8 @@ public class PollTableEntry extends AbstractPollTableEntry {
     private int maxRetryCount;
     private long reconnectTimeout;
     private boolean fileLocking;
+    private boolean autoLockRelease;
+    private Long autoLockReleaseInterval;
 
     private String moveAfterMoveFailure;
 
@@ -196,6 +198,22 @@ public boolean isFileLockingEnabled() {
         return fileLocking;
     }
 
+    public boolean getAutoLockRelease() {
+        return autoLockRelease;
+    }
+
+    public void setAutoLockRelease(boolean autoLockRelease) {
+        this.autoLockRelease = autoLockRelease;
+    }
+
+    public Long getAutoLockReleaseInterval() {
+        return autoLockReleaseInterval;
+    }
+
+    public void setAutoLockReleaseInterval(Long autoLockReleaseInterval) {
+        this.autoLockReleaseInterval = autoLockReleaseInterval;
+    }
+
     public long getReconnectTimeout() {
         return reconnectTimeout;
     }
@@ -281,6 +299,28 @@ public boolean loadConfiguration(ParameterInclude params) throws AxisFault {
                 fileLocking = false;
             }
 
+            String strAutoLockRelease = ParamUtils.getOptionalParam(
+                    params, VFSConstants.TRANSPORT_AUTO_LOCK_RELEASE);
+            autoLockRelease = false;
+            autoLockReleaseInterval = VFSConstants.DEFAULT_AUTO_LOCK_RELEASE_INTERVAL;
+            if (strAutoLockRelease != null) {
+                autoLockRelease = Boolean.parseBoolean(strAutoLockRelease);
+                if (autoLockRelease) {
+                    String strAutoLockReleaseInterval = ParamUtils.getOptionalParam(
+                            params, VFSConstants.TRANSPORT_AUTO_LOCK_RELEASE_INTERVAL);
+                    if (strAutoLockReleaseInterval != null) {
+                        try {
+                            autoLockReleaseInterval = Long.parseLong(strAutoLockReleaseInterval);
+                        } catch (NumberFormatException nfe) {
+                            log.warn("VFS " + VFSConstants.TRANSPORT_AUTO_LOCK_RELEASE_INTERVAL
+                                    + " value is invalid : " + strAutoLockReleaseInterval, nfe);
+                            autoLockReleaseInterval =
+                                    VFSConstants.DEFAULT_AUTO_LOCK_RELEASE_INTERVAL;
+                        }
+                    }
+                }
+            }
+
             moveAfterMoveFailure = ParamUtils.getOptionalParam(params,
                     VFSConstants.TRANSPORT_FILE_MOVE_AFTER_FAILED_MOVE);
 

modules/transports/core/vfs/src/main/java/org/apache/synapse/transport/vfs/VFSConstants.java

@@ -45,6 +45,23 @@ public final class VFSConstants {
     public static final String TRANSPORT_FILE_LOCKING_ENABLED = "enable";
     public static final String TRANSPORT_FILE_LOCKING_DISABLED = "disable";
 
+    /**
+     * Lock release true/false
+     * */
+    public static final String TRANSPORT_AUTO_LOCK_RELEASE = "transport.vfs.AutoLockRelease";
+
+    /**
+     * Lock release interval
+     * */
+    public static final String TRANSPORT_AUTO_LOCK_RELEASE_INTERVAL
+            = "transport.vfs.AutoLockReleaseInterval";
+
+    /**
+     * Age in milliseconds beyond which an existing lock file is treated as abandoned, used when
+     * auto lock release is enabled without an explicit interval.
+     */
+    public static final long DEFAULT_AUTO_LOCK_RELEASE_INTERVAL = 3600000L;
+
     public static final String REPLY_FILE_URI = "transport.vfs.ReplyFileURI";
     public static final String REPLY_FILE_NAME = "transport.vfs.ReplyFileName";
 

modules/transports/core/vfs/src/main/java/org/apache/synapse/transport/vfs/VFSParamDTO.java

@@ -0,0 +1,63 @@
+/*
+*  Licensed to the Apache Software Foundation (ASF) under one
+*  or more contributor license agreements.  See the NOTICE file
+*  distributed with this work for additional information
+*  regarding copyright ownership.  The ASF licenses this file
+*  to you under the Apache License, Version 2.0 (the
+*  "License"); you may not use this file except in compliance
+*  with the License.  You may obtain a copy of the License at
+*
+*   http://www.apache.org/licenses/LICENSE-2.0
+*
+*  Unless required by applicable law or agreed to in writing,
+*  software distributed under the License is distributed on an
+*   * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+*  KIND, either express or implied.  See the License for the
+*  specific language governing permissions and limitations
+*  under the License.
+*/
+package org.apache.synapse.transport.vfs;
+
+/**
+ * Carries the auto lock release settings of a poll table entry down to
+ * {@link VFSUtils#acquireLock(org.apache.commons.vfs2.FileSystemManager,
+ * org.apache.commons.vfs2.FileObject, VFSParamDTO)}.
+ */
+public class VFSParamDTO {
+
+    private boolean autoLockRelease;
+    private Long autoLockReleaseInterval;
+
+    public VFSParamDTO() {
+        autoLockRelease = false;
+        autoLockReleaseInterval = VFSConstants.DEFAULT_AUTO_LOCK_RELEASE_INTERVAL;
+    }
+
+    /**
+     * @return the autoLockRelease
+     */
+    public boolean isAutoLockRelease() {
+        return autoLockRelease;
+    }
+
+    /**
+     * @param autoLockRelease the autoLockRelease to set
+     */
+    public void setAutoLockRelease(boolean autoLockRelease) {
+        this.autoLockRelease = autoLockRelease;
+    }
+
+    /**
+     * @return the autoLockReleaseInterval
+     */
+    public Long getAutoLockReleaseInterval() {
+        return autoLockReleaseInterval;
+    }
+
+    /**
+     * @param autoLockReleaseInterval the autoLockReleaseInterval to set
+     */
+    public void setAutoLockReleaseInterval(Long autoLockReleaseInterval) {
+        this.autoLockReleaseInterval = autoLockReleaseInterval;
+    }
+}

modules/transports/core/vfs/src/main/java/org/apache/synapse/transport/vfs/VFSTransportListener.java

@@ -227,7 +227,7 @@ private void scanFileOrDirectory(final PollTableEntry entry, String fileURI) {
 
                     if (fileObject.getType() == FileType.FILE && !isFailedRecord) {
                         if (!entry.isFileLockingEnabled() || (entry.isFileLockingEnabled() &&
-                                VFSUtils.acquireLock(fsManager, fileObject))) {
+                                acquireLock(fsManager, fileObject, entry))) {
                             try {
                                 processFile(entry, fileObject);
                                 entry.setLastPollState(PollTableEntry.SUCCSESSFUL);
@@ -304,7 +304,7 @@ private void scanFileOrDirectory(final PollTableEntry entry, String fileURI) {
                             }
 
                             if ((!entry.isFileLockingEnabled()
-                                    || (entry.isFileLockingEnabled() && VFSUtils.acquireLock(fsManager, child)))
+                                    || (entry.isFileLockingEnabled() && acquireLock(fsManager, child, entry)))
                                     && !isFailedRecord){
                                 //process the file
                                 try {
@@ -395,6 +395,23 @@ private void scanFileOrDirectory(final PollTableEntry entry, String fileURI) {
         }
     }
 
+    /**
+     * Acquire the lock of the given file, passing the auto lock release settings configured on
+     * the poll table entry so that an abandoned lock does not keep the file unprocessed.
+     *
+     * @param fsManager used to resolve the lock file
+     * @param fileObject representing the file to be processed
+     * @param entry the PollTableEntry carrying the auto lock release configuration
+     * @return true if the lock has been acquired, false if not
+     */
+    protected boolean acquireLock(FileSystemManager fsManager, FileObject fileObject,
+                                  final PollTableEntry entry) {
+        VFSParamDTO vfsParamDTO = new VFSParamDTO();
+        vfsParamDTO.setAutoLockRelease(entry.getAutoLockRelease());
+        vfsParamDTO.setAutoLockReleaseInterval(entry.getAutoLockReleaseInterval());
+        return VFSUtils.acquireLock(fsManager, fileObject, vfsParamDTO);
+    }
+
     /**
      * Take specified action to either move or delete the processed file, depending on the outcome
      * @param entry the PollTableEntry for the file that has been processed

modules/transports/core/vfs/src/main/java/org/apache/synapse/transport/vfs/VFSUtils.java

@@ -24,6 +24,7 @@
 import org.apache.axis2.transport.base.BaseUtils;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.apache.commons.vfs2.Capability;
 import org.apache.commons.vfs2.FileContent;
 import org.apache.commons.vfs2.FileObject;
 import org.apache.commons.vfs2.FileSystemException;
@@ -111,6 +112,21 @@ public static String getFileName(MessageContext msgCtx, VFSOutTransportInfo vfsO
      * @return boolean true if the lock has been acquired or false if not
      */
     public synchronized static boolean acquireLock(FileSystemManager fsManager, FileObject fo) {
+        return acquireLock(fsManager, fo, null);
+    }
+
+    /**
+     * Acquires a file item lock before processing the item, guaranteeing that the file is not
+     * processed while it is being uploaded and/or the item is not processed by two listeners
+     *
+     * @param fsManager used to resolve the processing file
+     * @param fo representing the processing file item
+     * @param paramDTO the auto lock release settings of the poll table entry, or null to leave
+     *                 an existing lock in place regardless of its age
+     * @return boolean true if the lock has been acquired or false if not
+     */
+    public synchronized static boolean acquireLock(FileSystemManager fsManager, FileObject fo,
+                                                   VFSParamDTO paramDTO) {
 
         // generate a random lock value to ensure that there are no two parties
         // processing the same file
@@ -128,10 +144,16 @@ public synchronized static boolean acquireLock(FileSystemManager fsManager, File
             }
             FileObject lockObject = fsManager.resolveFile(fullPath + ".lock");
             if (lockObject.exists()) {
-                log.debug("There seems to be an external lock, aborting the processing of the file "
+                long lockAge = getLockAge(lockObject);
+                log.warn("There seems to be an external lock, aborting the processing of the file "
                         + maskURLPassword(fo.getName().getURI())
                         + ". This could possibly be due to some other party already "
-                        + "processing this file or the file is still being uploaded");
+                        + "processing this file or the file is still being uploaded"
+                        + (lockAge >= 0 ? ". The lock file is " + lockAge + "ms old" : ""));
+                if (paramDTO != null && paramDTO.isAutoLockRelease()
+                        && isLockAbandoned(lockAge, paramDTO.getAutoLockReleaseInterval())) {
+                    releaseLock(lockObject, lockAge);
+                }
             } else {
 
                 // write a lock file before starting of the processing, to ensure that the
@@ -196,6 +218,63 @@ public static void releaseLock(FileSystemManager fsManager, FileObject fo) {
         }
     }
 
+    /**
+     * Decide whether an existing lock file is old enough to be treated as abandoned.
+     *
+     * @param lockAge the age of the lock file, or -1 if it could not be determined
+     * @param autoLockReleaseInterval age in milliseconds beyond which the lock is abandoned
+     * @return true if the lock has not been updated for at least the configured interval
+     */
+    private static boolean isLockAbandoned(long lockAge, Long autoLockReleaseInterval) {
+        if (lockAge < 0) {
+            // the file system cannot report a last modified time, so the age of the lock is
+            // unknown; leave it in place rather than removing a lock that may still be held
+            return false;
+        }
+        return autoLockReleaseInterval == null || autoLockReleaseInterval <= lockAge;
+    }
+
+    /**
+     * Remove a lock file that {@link #isLockAbandoned} has established is no longer held.
+     *
+     * @param lockObject the existing lock file
+     * @param lockAge the age of the lock file, reported in the log
+     */
+    private static void releaseLock(FileObject lockObject, long lockAge) {
+        try {
+            log.warn("Removing the lock file "
+                    + maskURLPassword(lockObject.getName().getURI()) + " after " + lockAge
+                    + "ms without an update");
+            lockObject.delete();
+        } catch (Exception e) {
+            log.warn("Unable to delete the lock file during auto release cycle.", e);
+        } finally {
+            try {
+                lockObject.close();
+            } catch (FileSystemException e) {
+                log.error("Couldn't release the lock file "
+                        + maskURLPassword(lockObject.getName().getURI()), e);
+            }
+        }
+    }
+
+    /**
+     * @param lockObject the lock file whose age is required
+     * @return the age of the lock file in milliseconds, or -1 if the file system cannot report
+     *         a last modified time
+     */
+    private static long getLockAge(FileObject lockObject) {
+        try {
+            if (!lockObject.getFileSystem().hasCapability(Capability.GET_LAST_MODIFIED)) {
+                return -1;
+            }
+            return System.currentTimeMillis() - lockObject.getContent().getLastModifiedTime();
+        } catch (FileSystemException e) {
+            log.debug("Couldn't read the last modified time of the lock file", e);
+            return -1;
+        }
+    }
+
     /**
      * Mask the password of the connection url with ***
      * @param url the actual url