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);
        }
    }
@Test
    public void shouldSkipValueAndTimestampeInformationForErrorOnTimestampAndValueSerialization() throws ClassNotFoundException {
        final Class myClass = Class.forName("java.lang.String");
        final StateSerdes<Object, Object> stateSerdes =
            new StateSerdes<Object, Object>("anyName", Serdes.serdeFrom(myClass), new ValueAndTimestampSerde(Serdes.serdeFrom(myClass)));
        final Integer myInt = 123;
        final Exception e = assertThrows(StreamsException.class, () -> stateSerdes.rawValue(ValueAndTimestamp.make(myInt, 0L)));
        assertThat(
            e.getMessage(),
            equalTo(
                "A serializer (org.apache.kafka.common.serialization.StringSerializer) " +
                    "is not compatible to the actual value type (value type: java.lang.Integer). " +
                    "Change the default Serdes in StreamConfig or provide correct Serdes via method parameters."));
    }