public byte[] rawKey(final K key) {
        try {
            return keySerde.serializer().serialize(topic, key);
        } catch (final ClassCastException e) {
            final String keyClass = key == null ? "unknown because key is null" : key.getClass().getName();
            throw new StreamsException(
                    String.format("A serializer (%s) is not compatible to the actual key type " +
                                    "(key type: %s). Change the default Serdes in StreamConfig or " +
                                    "provide correct Serdes via method parameters.",
                            keySerializer().getClass().getName(),
                            keyClass),
                    e);
        }
    }
@Test
    public void shouldThrowIfIncompatibleSerdeForKey() throws ClassNotFoundException {
        final Class myClass = Class.forName("java.lang.String");
        final StateSerdes<Object, Object> stateSerdes = new StateSerdes<Object, Object>("anyName", Serdes.serdeFrom(myClass), Serdes.serdeFrom(myClass));
        final Integer myInt = 123;
        final Exception e = assertThrows(StreamsException.class, () -> stateSerdes.rawKey(myInt));
        assertThat(
            e.getMessage(),
            equalTo(
                "A serializer (org.apache.kafka.common.serialization.StringSerializer) " +
                    "is not compatible to the actual key type (key type: java.lang.Integer). " +
                    "Change the default Serdes in StreamConfig or provide correct Serdes via method parameters."));
    }