### Question:
StateDirectory { boolean lock(final TaskId taskId, int retry) throws IOException { final File lockFile; if (locks.containsKey(taskId)) { log.trace("{} Found cached state dir lock for task {}", logPrefix, taskId); return true; } try { lockFile = new File(directoryForTask(taskId), LOCK_FILE_NAME); } catch (ProcessorStateException e) { return false; } final FileChannel channel; try { channel = getOrCreateFileChannel(taskId, lockFile.toPath()); } catch (NoSuchFileException e) { return false; } final FileLock lock = tryLock(retry, channel); if (lock != null) { locks.put(taskId, lock); log.debug("{} Acquired state dir lock for task {}", logPrefix, taskId); } return lock != null; } StateDirectory(final String applicationId, final String stateDirConfig, final Time time);  StateDirectory(final String applicationId, final String threadId, final String stateDirConfig, final Time time); void cleanRemovedTasks(final long cleanupDelayMs);  }### Answer:
@Test public void shouldNotLockDeletedDirectory() throws Exception { final TaskId taskId = new TaskId(0, 0); Utils.delete(stateDir); assertFalse(directory.lock(taskId, 0)); }