@Override
    public Map<TopicPartition, Long> initialize() {
        final Set<String> storeNames = stateMgr.initialize();
        final Map<String, String> storeNameToTopic = topology.storeToChangelogTopic();
        for (final String storeName : storeNames) {
            final String sourceTopic = storeNameToTopic.get(storeName);
            final SourceNode<?, ?> source = topology.source(sourceTopic);
            deserializers.put(
                sourceTopic,
                new RecordDeserializer(
                    source,
                    deserializationExceptionHandler,
                    logContext,
                    droppedRecordsSensor(
                        Thread.currentThread().getName(),
                        processorContext.taskId().toString(),
                        processorContext.metrics()
                    )
                )
            );
        }
        initTopology();
        processorContext.initialize();
        lastFlush = time.milliseconds();
        return stateMgr.changelogOffsets();
    }
@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