MINOR: improve error message for Serde type miss match

streams/src/main/java/org/apache/kafka/streams/state/StateSerdes.java

@@ -21,6 +21,7 @@
 import org.apache.kafka.common.serialization.Serdes;
 import org.apache.kafka.common.serialization.Serializer;
 import org.apache.kafka.streams.errors.StreamsException;
+import org.apache.kafka.streams.state.internals.ValueAndTimestampSerializer;
 
 import java.util.Objects;
 
@@ -190,12 +191,20 @@ public byte[] rawValue(final V value) {
         try {
             return valueSerde.serializer().serialize(topic, value);
         } catch (final ClassCastException e) {
-            final String valueClass = value == null ? "unknown because value is null" : value.getClass().getName();
+            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 (value: %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.",
-                            valueSerializer().getClass().getName(),
+                            serializerClass.getName(),
                             valueClass),
                     e);
         }

streams/src/main/java/org/apache/kafka/streams/state/internals/ValueAndTimestampSerde.java

@@ -28,7 +28,7 @@ public class ValueAndTimestampSerde<V> implements Serde<ValueAndTimestamp<V>> {
     private final ValueAndTimestampSerializer<V> valueAndTimestampSerializer;
     private final ValueAndTimestampDeserializer<V> valueAndTimestampDeserializer;
 
-    ValueAndTimestampSerde(final Serde<V> valueSerde) {
+    public ValueAndTimestampSerde(final Serde<V> valueSerde) {
         Objects.requireNonNull(valueSerde);
         valueAndTimestampSerializer = new ValueAndTimestampSerializer<>(valueSerde.serializer());
         valueAndTimestampDeserializer = new ValueAndTimestampDeserializer<>(valueSerde.deserializer());

streams/src/main/java/org/apache/kafka/streams/state/internals/ValueAndTimestampSerializer.java

@@ -24,7 +24,7 @@
 import java.util.Map;
 import java.util.Objects;
 
-class ValueAndTimestampSerializer<V> implements Serializer<ValueAndTimestamp<V>> {
+public class ValueAndTimestampSerializer<V> implements Serializer<ValueAndTimestamp<V>> {
     public final Serializer<V> valueSerializer;
     private final Serializer<Long> timestampSerializer;
 

streams/src/test/java/org/apache/kafka/streams/state/StateSerdesTest.java

@@ -19,11 +19,16 @@
 import org.apache.kafka.common.serialization.Serdes;
 import org.apache.kafka.common.utils.Bytes;
 import org.apache.kafka.streams.errors.StreamsException;
+import org.apache.kafka.streams.state.internals.ValueAndTimestampSerde;
 import org.junit.Assert;
 import org.junit.Test;
 
 import java.nio.ByteBuffer;
 
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.equalTo;
+import static org.junit.Assert.assertThrows;
+
 @SuppressWarnings("unchecked")
 public class StateSerdesTest {
 
@@ -88,20 +93,47 @@ public void shouldThrowIfValueClassIsNull() {
         new StateSerdes<>("anyName", Serdes.ByteArray(), null);
     }
 
-    @Test(expected = StreamsException.class)
+    @Test
     public void shouldThrowIfIncompatibleSerdeForValue() 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;
-        stateSerdes.rawValue(myInt);
+        final Exception e = assertThrows(StreamsException.class, () -> stateSerdes.rawValue(myInt));
+        assertThat(
+            e.getMessage(),
+            equalTo(
+                "A serializer (value: 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."));
+    }
+
+    @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 (value: 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."));
     }
 
-    @Test(expected = StreamsException.class)
+    @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;
-        stateSerdes.rawKey(myInt);
+        final Exception e = assertThrows(StreamsException.class, () -> stateSerdes.rawKey(myInt));
+        assertThat(
+            e.getMessage(),
+            equalTo(
+                "A serializer (key: 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."));
     }
 
 }