code: public static FileLock takeOwnership(FileSystem fs, Path lockFile, LogEntry lastEntry, String spoutId)
            throws IOException {
      try {
        if(fs instanceof DistributedFileSystem ) {
          if( !((DistributedFileSystem) fs).recoverLease(lockFile) ) {
            LOG.warn("Unable to recover lease on lock file {} right now. Cannot transfer ownership. Will need to try later. Spout = {}", lockFile, spoutId);
            return null;
          }
        }
        return new FileLock(fs, lockFile, spoutId, lastEntry);
      } catch (IOException e) {
        if (e instanceof RemoteException &&
                ((RemoteException) e).unwrapRemoteException() instanceof AlreadyBeingCreatedException) {
          LOG.warn("Lock file " + lockFile + "is currently open. Cannot transfer ownership now. Will need to try later. Spout= " + spoutId, e);
          return null;
        } else { // unexpected error
          LOG.warn("Cannot transfer ownership now for lock file " + lockFile + ". Will need to try later. Spout =" + spoutId, e);
          throw e;
        }
      }
    }
docs: Takes ownership of the lock file if possible.
  @param lockFile
  @param lastEntry   last entry in the lock file. this param is an optimization.
  we dont scan the lock file again to find its last entry here since
  its already been done once by the logic used to check if the lock
  file is stale. so this value comes from that earlier scan.
  @param spoutId     spout id
  @throws IOException if unable to acquire
  @return null if lock File is not recoverable