private <K, V> void forwardInternal(final ProcessorNode<K, V, ?, ?> child,
                                        final Record<K, V> record) {
        setCurrentNode(child);

        child.process(record);

        if (child.isTerminalNode()) {
            streamTask.maybeRecordE2ELatency(record.timestamp(), currentSystemTimeMs(), child.name());
        }
    }
@throws StreamsException if an attempt is made to access this state store from an unknown node
@throws UnsupportedOperationException if the current streamTask type is standby
streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorContextImpl.java