org.apache.storm.hdfs.spout

Class FileLock

java.lang.Object

org.apache.storm.hdfs.spout.FileLock

public class FileLock
extends Object

Facility to synchronize access to HDFS files. Thread gains exclusive access to a file by acquiring a FileLock object. The lock itself is represented as file on HDFS. Relies on atomic file creation. Owning thread must heartbeat periodically on the lock to prevent the lock from being deemed as stale (i.e. lock whose owning thread have died).

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
static class	FileLock.LogEntry

Method Summary

Modifier and Type	Method and Description
static FileLock	acquireOldestExpiredLock(org.apache.hadoop.fs.FileSystem fs, org.apache.hadoop.fs.Path lockFilesDir, int locktimeoutSec, String spoutId) 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
static FileLock.LogEntry	getLastEntry(org.apache.hadoop.fs.FileSystem fs, org.apache.hadoop.fs.Path lockFile) returns the last log entry
static FileLock.LogEntry	getLastEntryIfStale(org.apache.hadoop.fs.FileSystem fs, org.apache.hadoop.fs.Path lockFile, long olderThan) checks if lockFile is older than ‘olderThan’ UTC time by examining the modification time on file and (if necessary) the timestamp in last log entry in the file.
FileLock.LogEntry	getLastLogEntry()
org.apache.hadoop.fs.Path	getLockFile()
void	heartbeat(String fileOffset)
static HdfsUtils.Pair<org.apache.hadoop.fs.Path,FileLock.LogEntry>	locateOldestExpiredLock(org.apache.hadoop.fs.FileSystem fs, org.apache.hadoop.fs.Path lockFilesDir, int locktimeoutSec) Finds 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
void	release() Release lock by deleting file
static FileLock	takeOwnership(org.apache.hadoop.fs.FileSystem fs, org.apache.hadoop.fs.Path lockFile, FileLock.LogEntry lastEntry, String spoutId) Takes ownership of the lock file if possible.
static FileLock	tryLock(org.apache.hadoop.fs.FileSystem fs, org.apache.hadoop.fs.Path fileToLock, org.apache.hadoop.fs.Path lockDirPath, String spoutId) returns lock on file or null if file is already locked.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Method Detail

tryLock

public static FileLock tryLock(org.apache.hadoop.fs.FileSystem fs,
                               org.apache.hadoop.fs.Path fileToLock,
                               org.apache.hadoop.fs.Path lockDirPath,
                               String spoutId)
                        throws IOException

returns lock on file or null if file is already locked. throws if unexpected problem

Throws:

IOException

getLastEntryIfStale

public static FileLock.LogEntry getLastEntryIfStale(org.apache.hadoop.fs.FileSystem fs,
                                                    org.apache.hadoop.fs.Path lockFile,
                                                    long olderThan)
                                             throws IOException

checks if lockFile is older than ‘olderThan’ UTC time by examining the modification time on file and (if necessary) the timestamp in last log entry in the file. If its stale, then returns the last log entry, else returns null.

Parameters:

fs -

lockFile -

olderThan - time (millis) in UTC.

Returns:

the last entry in the file if its too old. null if last entry is not too old

Throws:

IOException

getLastEntry

public static FileLock.LogEntry getLastEntry(org.apache.hadoop.fs.FileSystem fs,
                                             org.apache.hadoop.fs.Path lockFile)
                                      throws IOException

returns the last log entry

Parameters:

fs -

lockFile -

Returns:

Throws:

IOException

takeOwnership

public static FileLock takeOwnership(org.apache.hadoop.fs.FileSystem fs,
                                     org.apache.hadoop.fs.Path lockFile,
                                     FileLock.LogEntry lastEntry,
                                     String spoutId)
                              throws IOException

Takes ownership of the lock file if possible.

Parameters:

lockFile -

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.

spoutId - spout id

Returns:

null if lock File is not recoverable

Throws:

IOException - if unable to acquire

acquireOldestExpiredLock

public static FileLock acquireOldestExpiredLock(org.apache.hadoop.fs.FileSystem fs,
                                                org.apache.hadoop.fs.Path lockFilesDir,
                                                int locktimeoutSec,
                                                String spoutId)
                                         throws IOException

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

Parameters:

fs -

lockFilesDir -

locktimeoutSec -

Returns:

Throws:

IOException

locateOldestExpiredLock

public static HdfsUtils.Pair<org.apache.hadoop.fs.Path,FileLock.LogEntry> locateOldestExpiredLock(org.apache.hadoop.fs.FileSystem fs,
                                                                                                  org.apache.hadoop.fs.Path lockFilesDir,
                                                                                                  int locktimeoutSec)
                                                                                           throws IOException

Finds 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

Parameters:

fs -

lockFilesDir -

locktimeoutSec -

Returns:

a Pair<lock file path, last entry in lock file> .. if expired lock file found

Throws:

IOException

heartbeat

public void heartbeat(String fileOffset)
               throws IOException

Throws:

IOException

release

public void release()
             throws IOException

Release lock by deleting file

Throws:

IOException - if lock file could not be deleted

getLastLogEntry

public FileLock.LogEntry getLastLogEntry()

getLockFile

public org.apache.hadoop.fs.Path getLockFile()