"WindowedBy Count KStream" throws StreamsException

Question

I tried to count event from KStream, into time period:

    KStream<String, VehicleEventTO> stream = builder.stream("vehicle", Consumed.with(Serdes.String(), new JsonSerde<>(VehicleEventTO.class)));

    KStream<String, VehicleEventTO> streamWithKey = stream.selectKey((key, value) -> value.getId_vehicle().toString());

    KStream<String, Long> streamValueKey = streamWithKey.map((key, value) -> KeyValue.pair(key, value.getId_vehicle()));

    streamValueKey.groupByKey()
                  .windowedBy(TimeWindows.of(Duration.ofMinutes(10).toMillis()))
                  .count(Materialized.with(Serdes.String(), new JsonSerde<>(Long.class)));

I've this exception:

Exception in thread "test-app-87ce164d-c427-4dcf-aa76-aeeb6f8fc943-StreamThread-1" org.apache.kafka.streams.errors.StreamsException: Exception caught in process. taskId=0_0, processor=KSTREAM-SOURCE-0000000000, topic=vehicle, partition=0, offset=160385 at org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:318) at org.apache.kafka.streams.processor.internals.AssignedStreamsTasks.process(AssignedStreamsTasks.java:94) at org.apache.kafka.streams.processor.internals.TaskManager.process(TaskManager.java:409) at org.apache.kafka.streams.processor.internals.StreamThread.processAndMaybeCommit(StreamThread.java:964) at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:832) at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:767) at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:736) Caused by: org.apache.kafka.streams.errors.StreamsException: A serializer (key: org.apache.kafka.common.serialization.ByteArraySerializer / value: org.apache.kafka.common.serialization.ByteArraySerializer) is not compatible to the actual key or value type (key type: java.lang.String / value type: java.lang.Long). Change the default Serdes in StreamConfig or provide correct Serdes via method parameters.

Giorgos Myrianthous Giorgos Myrianthous · Accepted Answer · 2019-07-01T10:32:09

groupByKey() makes use of the default serialisers:

groupByKey()

Group the records by their current key into a KGroupedStream while preserving the original values and default serializers and deserializers.

You either have to use groupByKey(Serialized<K,V> serialized) or groupByKey(Grouped<K,V> grouped).

The following should do the trick:

streamValueKey.groupByKey(Serialized.with(Serdes.String(), Serdes.Long()))
              .windowedBy(TimeWindows.of(Duration.ofMinutes(10).toMillis()))
              .count(Materialized.with(Serdes.String(), new JsonSerde<>(Long.class)));

"WindowedBy Count KStream" throws StreamsException

1 Answers