public void close(final boolean wipeStateStore) throws IOException {
        stateMgr.close();
        if (wipeStateStore) {
            try {
                log.info("Deleting global task directory after detecting corruption.");
                Utils.delete(stateMgr.baseDir());
            } catch (final IOException e) {
                log.error("Failed to delete global task directory after detecting corruption.", e);
            }
        }
    }
@throws IllegalStateException If store gets registered after initialized is already finished
@throws StreamsException      If the store's change log does not contain the partition
streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateUpdateTask.java