class ProcessorNode {

    @SuppressWarnings("unchecked")
    public void init(final InternalProcessorContext context) {
        if (!closed)
            throw new IllegalStateException("The processor is not closed");

        try {
            internalProcessorContext = context;
            initSensors();
            maybeMeasureLatency(
                () -> {
                    if (processor != null) {
                        processor.init((ProcessorContext<KOut, VOut>) context);
                    }
                },
                time,
                createSensor
            );
        } catch (final Exception e) {
            throw new StreamsException(String.format("failed to initialize processor %s", name), e);
        }

        // revived tasks could re-initialize the topology,
        // in which case we should reset the flag
        closed = false;
    }

    public  ProcessorNode(final String name);
    public  ProcessorNode(final String name,
                         final Processor<KIn, VIn, KOut, VOut> processor,
                         final Set<String> stateStores);
    public  ProcessorNode(final String name,
                         final org.apache.kafka.streams.processor.Processor<KIn, VIn> processor,
                         final Set<String> stateStores);

}