AbstractStreamedMetricProcessorSupplier.java 文件源码

/**
 * {@inheritDoc}
 * @see org.apache.kafka.streams.processor.ProcessorSupplier#get()
 */
@Override
public Processor<K, V> get() {
    final AbstractStreamedMetricProcessor<K,V> processor = getProcessor(topicSink, sources);

    startedProcessors.add(processor);
    final String processorBeanName = processor.getClass().getSimpleName() + "#" + processor.getInstanceId(); 
    processor.setBeanName(processorBeanName);

    ((SingletonBeanRegistry)appCtx.getAutowireCapableBeanFactory()).registerSingleton(processorBeanName, processor);
    appCtx.getAutowireCapableBeanFactory().autowireBean(processor);
    return processor;
}