@Override
    public void init(final InternalProcessorContext<KIn, VIn> context) {
        // It is important to first create the sensor before calling init on the
        // parent object. Otherwise due to backwards compatibility an empty sensor
        // without parent is created with the same name.
        // Once the backwards compatibility is not needed anymore it might be possible to
        // change this.
        processAtSourceSensor = ProcessorNodeMetrics.processAtSourceSensor(
            Thread.currentThread().getName(),
            context.taskId().toString(),
            context.currentNode().name(),
            context.metrics()
        );
        super.init(context);
        this.context = context;

        try {
            keyDeserializer = prepareKeyDeserializer(keyDeserializer, context, name());
        } catch (final ConfigException | StreamsException e) {
            throw new StreamsException(String.format("Failed to initialize key serdes for source node %s", name()), e, context.taskId());
        }

        try {
            valDeserializer = prepareValueDeserializer(valDeserializer, context, name());
        } catch (final ConfigException | StreamsException e) {
            throw new StreamsException(String.format("Failed to initialize value serdes for source node %s", name()), e, context.taskId());
        }
    }
@Test
    public void shouldThrowStreamsExceptionOnUndefinedKeySerde() {
        final InternalMockProcessorContext<String, String> context = new InternalMockProcessorContext<>();

        final SourceNode<String, String> node =
            new SourceNode<>(context.currentNode().name(), new TheDeserializer(), new TheDeserializer());

        utilsMock.when(() -> WrappingNullableUtils.prepareKeyDeserializer(any(), any(), any()))
            .thenThrow(new ConfigException("Please set StreamsConfig#DEFAULT_KEY_SERDE_CLASS_CONFIG"));

        final Throwable exception = assertThrows(StreamsException.class, () -> node.init(context));

        assertThat(
            exception.getMessage(),
            equalTo("Failed to initialize key serdes for source node TESTING_NODE")
        );
        assertThat(
            exception.getCause().getMessage(),
            equalTo("Please set StreamsConfig#DEFAULT_KEY_SERDE_CLASS_CONFIG")
        );
    }