static <K, V> StateSerdes<K, V> prepareStoreSerde(final StateStoreContext context,
                                                      final String storeName,
                                                      final String changelogTopic,
                                                      final Serde<K> keySerde,
                                                      final Serde<V> valueSerde,
                                                      final PrepareFunc<V> prepareValueSerdeFunc) {
        return new StateSerdes<>(
            changelogTopic,
            prepareSerde(WrappingNullableUtils::prepareKeySerde, storeName, keySerde, new SerdeGetter(context), true, context.taskId()),
            prepareSerde(prepareValueSerdeFunc, storeName, valueSerde, new SerdeGetter(context), false, context.taskId())
        );
    }
@Test
    public void shouldThrowStreamsExceptionOnUndefinedKeySerdeForStateStoreContext() {
        final MockInternalNewProcessorContext<String, String> context = new MockInternalNewProcessorContext<>();

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

        final Throwable exception = assertThrows(StreamsException.class,
            () -> StoreSerdeInitializer.prepareStoreSerde((StateStoreContext) context, "myStore", "topic",
                new Serdes.StringSerde(), new Serdes.StringSerde(), WrappingNullableUtils::prepareValueSerde));

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