kafka streams example throws a class cast exception, serde windowed -> string. How to set the proper Serde?

Viewed 632

I'm attempting to reproduce this example. My topology is:

@Bean("myTopo")
    public KStream<Object, Object> getTopo(@Qualifier("myKConfig") StreamsBuilder builder) {
        var stream = builder.stream("my-events");
        stream.groupByKey()
                .windowedBy(TimeWindows.of(Duration.ofMinutes(2)))
                .count()
                .suppress(Suppressed.untilWindowCloses(Suppressed.BufferConfig.unbounded()))
                .toStream()
                .foreach((k, v) -> {
                    System.out.println("k + v = " + k + " --- " + v);
                });

I've set the serde and the windowed serde internal classes in the config:

        ...
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, JsonSerde.class);
        ...
        props.put(JsonDeserializer.VALUE_DEFAULT_TYPE, JsonNode.class);
        props.put(StreamsConfig.DEFAULT_WINDOWED_KEY_SERDE_INNER_CLASS, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_WINDOWED_VALUE_SERDE_INNER_CLASS, JsonSerde.class);
        var config = new KafkaStreamsConfiguration(props);
        return new StreamsBuilderFactoryBean(config);

The error I get is

org.apache.kafka.streams.errors.StreamsException: ClassCastException invoking Processor. 
Do the Processor's input types match the deserialized types? 
Check the Serde setup and change the default Serdes in StreamConfig or provide correct Serdes via method parameters. 
Make sure the Processor can accept the deserialized input of type key: 
   org.apache.kafka.streams.kstream.Windowed, 
and value: 
   org.apache.kafka.streams.kstream.internals.Change.

with underlying cause

java.lang.ClassCastException: class org.apache.kafka.streams.kstream.Windowed 
cannot be cast to class java.lang.String (org.apache.kafka.streams.kstream.Windowed is in unnamed module of loader 'app'; 
java.lang.String is in module java.base of loader 'bootstrap')

I see count() returns KTable<Windowed<Object>, Long>. So it looks like the problem is it wants a Windowed<String> serde for the key. Apparently, DEFAULT_WINDOWED_KEY_SERDE_INNER_CLASS isn't sufficient.

How do I create and set this?

1 Answers

I think I ran into this bug:

https://issues.apache.org/jira/browse/KAFKA-9259

I added a Materialized to the count() method

        var store = Stores.persistentTimestampedWindowStore(
                "some-state-store",
                Duration.ofMinutes(5),
                Duration.ofMinutes(2),
                false);
        var materialized = Materialized
                .<String, Long>as(store)
                .withKeySerde(Serdes.String());

Now the code runs without exception.

Related