@Override
    public void process(final Record<KIn, VIn> record) {
        final RecordCollector collector = ((RecordCollector.Supplier) context).recordCollector();

        final KIn key = record.key();
        final VIn value = record.value();

        final long timestamp = record.timestamp();

        final ProcessorRecordContext contextForExtraction =
            new ProcessorRecordContext(
                timestamp,
                context.offset(),
                context.partition(),
                context.topic(),
                record.headers(),
                context.recordContext().rawRecord()
            );

        final String topic = topicExtractor.extract(key, value, contextForExtraction);

        collector.send(
            topic,
            key,
            value,
            record.headers(),
            timestamp,
            keySerializer,
            valSerializer,
            name(),
            context,
            partitioner);
    }
@throws UnsupportedOperationException if this method adds a child to a sink node
streams/src/main/java/org/apache/kafka/streams/processor/internals/SinkNode.java