@Override
    public void close() {
        if (hasPersistentStores) {
            try {
                stateDirLock.release();
                stateDirLockChannel.close();

                stateDirLock = null;
                stateDirLockChannel = null;
            } catch (final IOException e) {
                log.error("Unexpected exception while unlocking the state dir", e);
                throw new StreamsException(String.format("Failed to release the lock on the state directory [%s]", stateDir.getAbsolutePath()), e);
            }

            // all threads should be stopped and cleaned up by now, so none should remain holding a lock
            if (!lockedTasksToOwner.isEmpty()) {
                log.error("Some task directories still locked while closing state, this indicates unclean shutdown: {}", lockedTasksToOwner);
            }
        }
    }
Unlock the state directory for the given {@link TaskId}.
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java