FileLock { public void heartbeat(String fileOffset) throws IOException { logProgress(fileOffset, true); } private  FileLock(FileSystem fs, Path lockFile, FSDataOutputStream lockFileStream, String spoutId); private  FileLock(FileSystem fs, Path lockFile, String spoutId, LogEntry entry); void heartbeat(String fileOffset); void release(); static FileLock tryLock(FileSystem fs, Path fileToLock, Path lockDirPath, String spoutId); static LogEntry getLastEntryIfStale(FileSystem fs, Path lockFile, long olderThan); static LogEntry getLastEntry(FileSystem fs, Path lockFile); static FileLock takeOwnership(FileSystem fs, Path lockFile, LogEntry lastEntry, String spoutId); static FileLock acquireOldestExpiredLock(FileSystem fs, Path lockFilesDir, int locktimeoutSec, String spoutId); static HdfsUtils.Pair<Path,LogEntry> locateOldestExpiredLock(FileSystem fs, Path lockFilesDir, int locktimeoutSec); LogEntry getLastLogEntry(); Path getLockFile();  }
@Test public void testHeartbeat() throws Exception { Path file1 = new Path(filesDir + Path.SEPARATOR + "file1"); fs.create(file1).close(); FileLock lock1 = FileLock.tryLock(fs, file1, locksDir, "spout1"); Assert.assertNotNull(lock1); Assert.assertTrue(fs.exists(lock1.getLockFile())); ArrayList<String> lines = readTextFile(lock1.getLockFile()); Assert.assertEquals("heartbeats appear to be missing", 1, lines.size()); lock1.heartbeat("1"); lock1.heartbeat("2"); lock1.heartbeat("3"); lines = readTextFile(lock1.getLockFile()); Assert.assertEquals("heartbeats appear to be missing", 4, lines.size()); lock1.heartbeat("4"); lock1.heartbeat("5"); lock1.heartbeat("6"); lines = readTextFile(lock1.getLockFile()); Assert.assertEquals("heartbeats appear to be missing", 7, lines.size()); lock1.release(); lines = readTextFile(lock1.getLockFile()); Assert.assertNull(lines); Assert.assertFalse(fs.exists(lock1.getLockFile())); }