class StateDirectory {

    synchronized void unlock(final TaskId taskId) throws IOException {
        final LockAndOwner lockAndOwner = locks.get(taskId);
        if (lockAndOwner != null && lockAndOwner.owningThread.equals(Thread.currentThread().getName())) {
            locks.remove(taskId);
            lockAndOwner.lock.release();
            log.debug("{} Released state dir lock for task {}", logPrefix(), taskId);

            final FileChannel fileChannel = channels.remove(taskId);
            if (fileChannel != null) {
                fileChannel.close();
            }
        }
    }

    public  StateDirectory(final StreamsConfig config, final Time time, final boolean hasPersistentStores);

}