@SuppressWarnings("unchecked")
    private void initializeTopology() {
        // initialize the task by initializing all its processor nodes in the topology
        log.trace("Initializing processor nodes of the topology");
        for (final ProcessorNode<?, ?, ?, ?> node : topology.processors()) {
            processorContext.setCurrentNode(node);
            try {
                node.init(processorContext, processingExceptionHandler);
            } finally {
                processorContext.setCurrentNode(null);
            }
        }
    }
Return all the checkpointable offsets(written + consumed) to the state manager.
Currently only changelog topic offsets need to be checkpointed.
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java