private static String getPositionString(final TopicPartition partition,
                                            final ChangelogMetadata changelogMetadata) {
        final ProcessorStateManager stateManager = changelogMetadata.stateManager;
        final Long offsets = stateManager.changelogOffsets().get(partition);
        return offsets == null ? "unknown" : String.valueOf(offsets);
    }
Since it is shared for multiple tasks and hence multiple state managers, the registration would take its
corresponding state manager as well for restoring.
streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java