@Test
public void testLockStateDirectory() throws IOException {

    File baseDir = Files.createTempDirectory(stateDir).toFile();
    try {
        FileLock lock;
        // the state manager locks the directory
        ProcessorStateManager stateMgr = new ProcessorStateManager(applicationId, 1, noPartitions, baseDir, new MockRestoreConsumer(), false);
        try {
            // this should not get the lock
            lock = ProcessorStateManager.lockStateDirectory(baseDir);
            assertNull(lock);
        } finally {
            // by closing the state manager, release the lock
            stateMgr.close(Collections.<TopicPartition, Long>emptyMap());
        }
        // now, this should get the lock
        lock = ProcessorStateManager.lockStateDirectory(baseDir);
        try {
            assertNotNull(lock);
        } finally {
            if (lock != null) {
                lock.release();
                lock.channel().close();
            }
        }
    } finally {
        Utils.delete(baseDir);
    }
}