@Override
    public void init(final InternalProcessorContext<Void, Void> context) {
        super.init(context);
        this.context = context;
        try {
            keySerializer = prepareKeySerializer(keySerializer, context, this.name());
        } catch (ConfigException | StreamsException e) {
            throw new StreamsException(String.format("Failed to initialize key serdes for sink node %s", name()), e, context.taskId());
        }

        try {
            valSerializer = prepareValueSerializer(valSerializer, context, this.name());
        } catch (final ConfigException | StreamsException e) {
            throw new StreamsException(String.format("Failed to initialize value serdes for sink node %s", name()), e, context.taskId());
        }
    }
@Test
    public void shouldThrowStreamsExceptionOnUndefinedKeySerde() {
        utilsMock.when(() -> WrappingNullableUtils.prepareKeySerializer(any(), any(), any()))
            .thenThrow(new ConfigException("Please set StreamsConfig#DEFAULT_KEY_SERDE_CLASS_CONFIG"));

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

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