class SinkNode extends ProcessorNode<KIn, VIn, KOut, VOut> {

    @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()
            );

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

        collector.send(topic, key, value, record.headers(), timestamp, keySerializer, valSerializer, partitioner);
    }

      SinkNode(final String name,
             final TopicNameExtractor<KIn, VIn> topicExtractor,
             final Serializer<KIn> keySerializer,
             final Serializer<VIn> valSerializer,
             final StreamPartitioner<? super KIn, ? super VIn> partitioner);

}