@Override
    public void updateChangelogOffsets(final Map<TopicPartition, Long> writtenOffsets) {
        for (final Map.Entry<TopicPartition, Long> entry : writtenOffsets.entrySet()) {
            final StateStoreMetadata store = findStore(entry.getKey());

            if (store != null) {
                store.setOffset(entry.getValue());

                log.debug("State store {} updated to written offset {} at changelog {}",
                        store.stateStore.name(), store.offset, store.changelogPartition);
            }
        }
    }
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