public void init(final InternalProcessorContext<KOut, VOut> context) {
        if (!closed)
            throw new IllegalStateException("The processor is not closed");

        try {
            threadId = Thread.currentThread().getName();
            internalProcessorContext = context;
            droppedRecordsSensor = TaskMetrics.droppedRecordsSensor(threadId,
                internalProcessorContext.taskId().toString(),
                internalProcessorContext.metrics());

            if (processor != null) {
                processor.init(context);
            }
            if (fixedKeyProcessor != null) {
                @SuppressWarnings("unchecked") final FixedKeyProcessorContext<KIn, VOut> fixedKeyProcessorContext =
                    (FixedKeyProcessorContext<KIn, VOut>) context;
                fixedKeyProcessor.init(fixedKeyProcessorContext);
            }
        } 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;
    }
@Test
    public void shouldThrowStreamsExceptionIfExceptionCaughtDuringClose() {
        final ProcessorNode<Object, Object, Object, Object> node =
            new ProcessorNode<>(NAME, new ExceptionalProcessor(), Collections.emptySet());
        assertThrows(StreamsException.class, () -> node.init(null));
    }