code: public static FileLock acquireOldestExpiredLock(FileSystem fs, Path lockFilesDir, int locktimeoutSec, String spoutId)
            throws IOException {
      // list files
      long now = System.currentTimeMillis();
      long olderThan = now - (locktimeoutSec*1000);
      Collection<Path> listing = HdfsUtils.listFilesByModificationTime(fs, lockFilesDir, olderThan);
  
      // locate expired lock files (if any). Try to take ownership (oldest lock first)
      for (Path file : listing) {
        if(file.getName().equalsIgnoreCase( DirLock.DIR_LOCK_FILE) ) {
          continue;
        }
        LogEntry lastEntry = getLastEntryIfStale(fs, file, olderThan);
        if(lastEntry!=null) {
          FileLock lock = FileLock.takeOwnership(fs, file, lastEntry, spoutId);
          if(lock!=null) {
            return lock;
          }
        }
      }
      if(listing.isEmpty()) {
        LOG.debug("No abandoned lock files found by Spout {}", spoutId);
      }
      return null;
    }
docs: Finds a oldest expired lock file (using modification timestamp), then takes
  ownership of the lock file
  Impt: Assumes access to lockFilesDir has been externally synchronized such that
  only one thread accessing the same thread
  @param fs
  @param lockFilesDir
  @param locktimeoutSec
  @return