@SuppressWarnings("unchecked")
    private void initTopology() {
        for (final ProcessorNode<?, ?, ?, ?> node : this.topology.processors()) {
            processorContext.setCurrentNode(node);
            try {
                node.init(this.processorContext);
            } finally {
                processorContext.setCurrentNode(null);
            }
        }
    }
@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