void recycle() {
        log.debug("Recycling state for {} task {}.", taskType, taskId);

        if (!stateUpdaterEnabled) {
            final List<TopicPartition> allChangelogs = getAllChangelogTopicPartitions();
            changelogReader.unregister(allChangelogs);
        }

        // when the state manager is recycled to be used, future writes may bypass its store's caching
        // layer if they are from restoration, hence we need to clear the state store's caches just in case
        // See KAFKA-14172 for details
        if (!stores.isEmpty()) {
            log.debug("Clearing all store caches registered in the state manager: {}", stores);
            for (final StateStoreMetadata metadata : stores.values()) {
                final StateStore store = metadata.stateStore;

                if (store instanceof CachedStateStore) {
                    ((CachedStateStore) store).clearCache();
                }
                log.trace("Cleared cache {}", store.name());
            }
        }
    }
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