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);
        }
    }
Serialize the given key.

@param key  the key to be serialized
@return     the serialized key
streams/src/main/java/org/apache/kafka/streams/state/StateSerdes.java