class StateSerdes {

    public byte[] rawValue(final V value) {
        try {
            return valueSerde.serializer().serialize(topic, value);
        } catch (final ClassCastException e) {
            final String valueClass;
            final Class<? extends Serializer> serializerClass;
            if (valueSerializer() instanceof ValueAndTimestampSerializer) {
                serializerClass = ((ValueAndTimestampSerializer) valueSerializer()).valueSerializer.getClass();
                valueClass = value == null ? "unknown because value is null" : ((ValueAndTimestamp) value).value().getClass().getName();
            } else {
                serializerClass = valueSerializer().getClass();
                valueClass = value == null ? "unknown because value is null" : value.getClass().getName();
            }
            throw new StreamsException(
                    String.format("A serializer (%s) is not compatible to the actual value type " +
                                    "(value type: %s). Change the default Serdes in StreamConfig or " +
                                    "provide correct Serdes via method parameters.",
                            serializerClass.getName(),
                            valueClass),
                    e);
        }
    }

    public  StateSerdes(final String topic,
                       final Serde<K> keySerde,
                       final Serde<V> valueSerde);

}