@Override
    public void checkpoint() {
        // checkpoint those stores that are only logged and persistent to the checkpoint file
        final Map<TopicPartition, Long> checkpointingOffsets = new HashMap<>();
        for (final StateStoreMetadata storeMetadata : stores.values()) {
            if (storeMetadata.commitCallback != null && !storeMetadata.corrupted) {
                try {
                    storeMetadata.commitCallback.onCommit();
                } catch (final IOException e) {
                    throw new ProcessorStateException(
                            format("%sException caught while trying to checkpoint store, " +
                                    "changelog partition %s", logPrefix, storeMetadata.changelogPartition),
                            e
                    );
                }
            }

            // store is logged, persistent, not corrupted, and has a valid current offset
            if (storeMetadata.changelogPartition != null &&
                storeMetadata.stateStore.persistent() &&
                !storeMetadata.corrupted) {

                final long checkpointableOffset = checkpointableOffsetFromChangelogOffset(storeMetadata.offset);
                checkpointingOffsets.put(storeMetadata.changelogPartition, checkpointableOffset);
            }
        }

        log.debug("Writing checkpoint: {} for task {}", checkpointingOffsets, taskId);
        try {
            checkpointFile.write(checkpointingOffsets);
        } catch (final IOException e) {
            log.warn("Failed to write offset checkpoint file to [{}]." +
                " This may occur if OS cleaned the state.dir in case when it located in ${java.io.tmpdir} directory." +
                " This may also occur due to running multiple instances on the same machine using the same state dir." +
                " Changing the location of state.dir may resolve the problem.",
                checkpointFile, e);
        }
    }
Alternative to {@link #close()} that just resets the changelogs without closing any of the underlying state
or unregistering the stores themselves
streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java